From b1ba97c4f544e5f71a3531c8e8253b86532f3dc8 Mon Sep 17 00:00:00 2001 From: Jerry Date: Tue, 5 Jul 2022 17:32:05 -0700 Subject: [PATCH] Create MVHashMap and use it StateDB --- core/blockstm/mvhashmap.go | 218 +++++++++++++++++++ core/blockstm/txio.go | 75 +++++++ core/state/journal.go | 8 + core/state/statedb.go | 436 ++++++++++++++++++++++++++++++------- core/state/statedb_test.go | 424 ++++++++++++++++++++++++++++++++++++ go.mod | 2 + go.sum | 2 + 7 files changed, 1085 insertions(+), 80 deletions(-) create mode 100644 core/blockstm/mvhashmap.go create mode 100644 core/blockstm/txio.go diff --git a/core/blockstm/mvhashmap.go b/core/blockstm/mvhashmap.go new file mode 100644 index 0000000000..ff9f1a6f9d --- /dev/null +++ b/core/blockstm/mvhashmap.go @@ -0,0 +1,218 @@ +package blockstm + +import ( + "fmt" + "sync" + + "github.com/emirpasic/gods/maps/treemap" +) + +const FlagDone = 0 +const FlagEstimate = 1 + +type MVHashMap struct { + rw sync.RWMutex + m map[string]*TxnIndexCells // TODO: might want a more efficient key representation +} + +func MakeMVHashMap() *MVHashMap { + return &MVHashMap{ + rw: sync.RWMutex{}, + m: make(map[string]*TxnIndexCells), + } +} + +type WriteCell struct { + flag uint + incarnation int + data interface{} +} + +type TxnIndexCells struct { + rw sync.RWMutex + tm *treemap.Map +} + +type Version struct { + TxnIndex int + Incarnation int +} + +func (mv *MVHashMap) getKeyCells(k []byte, fNoKey func(kenc string) *TxnIndexCells) (cells *TxnIndexCells) { + kenc := string(k) + + var ok bool + + mv.rw.RLock() + cells, ok = mv.m[kenc] + mv.rw.RUnlock() + + if !ok { + cells = fNoKey(kenc) + } + + return +} + +func (mv *MVHashMap) Write(k []byte, v Version, data interface{}) { + cells := mv.getKeyCells(k, func(kenc string) (cells *TxnIndexCells) { + n := &TxnIndexCells{ + rw: sync.RWMutex{}, + tm: treemap.NewWithIntComparator(), + } + var ok bool + mv.rw.Lock() + if cells, ok = mv.m[kenc]; !ok { + mv.m[kenc] = n + cells = n + } + mv.rw.Unlock() + return + }) + + // TODO: could probably have a scheme where this only generally requires a read lock since any given transaction transaction + // should only have one incarnation executing at a time... + cells.rw.Lock() + defer cells.rw.Unlock() + ci, ok := cells.tm.Get(v.TxnIndex) + + if ok { + if ci.(*WriteCell).incarnation > v.Incarnation { + panic(fmt.Errorf("existing transaction value does not have lower incarnation: %v, %v", + string(k), v.TxnIndex)) + } else if ci.(*WriteCell).flag == FlagEstimate { + println("marking previous estimate as done tx", v.TxnIndex, v.Incarnation) + } + + ci.(*WriteCell).flag = FlagDone + ci.(*WriteCell).incarnation = v.Incarnation + ci.(*WriteCell).data = data + } else { + cells.tm.Put(v.TxnIndex, &WriteCell{ + flag: FlagDone, + incarnation: v.Incarnation, + data: data, + }) + } +} + +func (mv *MVHashMap) MarkEstimate(k []byte, txIdx int) { + cells := mv.getKeyCells(k, func(_ string) *TxnIndexCells { + panic(fmt.Errorf("path must already exist")) + }) + + cells.rw.RLock() + if ci, ok := cells.tm.Get(txIdx); !ok { + panic("should not happen - cell should be present for path") + } else { + ci.(*WriteCell).flag = FlagEstimate + } + cells.rw.RUnlock() +} + +func (mv *MVHashMap) Delete(k []byte, txIdx int) { + cells := mv.getKeyCells(k, func(_ string) *TxnIndexCells { + panic(fmt.Errorf("path must already exist")) + }) + + cells.rw.Lock() + defer cells.rw.Unlock() + cells.tm.Remove(txIdx) +} + +const ( + MVReadResultDone = 0 + MVReadResultDependency = 1 + MVReadResultNone = 2 +) + +type MVReadResult struct { + depIdx int + incarnation int + value interface{} +} + +func (res *MVReadResult) DepIdx() int { + return res.depIdx +} + +func (res *MVReadResult) Incarnation() int { + return res.incarnation +} + +func (res *MVReadResult) Value() interface{} { + return res.value +} + +func (mvr MVReadResult) Status() int { + if mvr.depIdx != -1 { + if mvr.incarnation == -1 { + return MVReadResultDependency + } else { + return MVReadResultDone + } + } + + return MVReadResultNone +} + +func (mv *MVHashMap) Read(k []byte, txIdx int) (res MVReadResult) { + res.depIdx = -1 + res.incarnation = -1 + + cells := mv.getKeyCells(k, func(_ string) *TxnIndexCells { + return nil + }) + if cells == nil { + return + } + + cells.rw.RLock() + defer cells.rw.RUnlock() + + if fk, fv := cells.tm.Floor(txIdx - 1); fk != nil && fv != nil { + c := fv.(*WriteCell) + switch c.flag { + case FlagEstimate: + res.depIdx = fk.(int) + res.value = c.data + case FlagDone: + { + res.depIdx = fk.(int) + res.incarnation = c.incarnation + res.value = c.data + } + default: + panic(fmt.Errorf("should not happen - unknown flag value")) + } + } + + return +} + +func ValidateVersion(txIdx int, lastInputOutput *TxnInputOutput, versionedData *MVHashMap) (valid bool) { + valid = true + + for _, rd := range lastInputOutput.readSet(txIdx) { + mvResult := versionedData.Read(rd.Path, txIdx) + switch mvResult.Status() { + case MVReadResultDone: + valid = rd.Kind == ReadKindMap && rd.V == Version{ + TxnIndex: mvResult.depIdx, + Incarnation: mvResult.incarnation, + } + case MVReadResultDependency: + valid = false + case MVReadResultNone: + valid = rd.Kind == ReadKindStorage // feels like an assertion? + default: + panic(fmt.Errorf("should not happen - undefined mv read status: %ver", mvResult.Status())) + } + + if !valid { + break + } + } + + return +} diff --git a/core/blockstm/txio.go b/core/blockstm/txio.go new file mode 100644 index 0000000000..4325277c1d --- /dev/null +++ b/core/blockstm/txio.go @@ -0,0 +1,75 @@ +//nolint: unused +package blockstm + +import "encoding/base64" + +const ( + ReadKindMap = 0 + ReadKindStorage = 1 +) + +type ReadDescriptor struct { + Path []byte + Kind int + V Version +} + +type WriteDescriptor struct { + Path []byte + V Version + Val interface{} +} + +type TxnInput []ReadDescriptor +type TxnOutput []WriteDescriptor + +// hasNewWrite: returns true if the current set has a new write compared to the input +func (txo TxnOutput) hasNewWrite(cmpSet []WriteDescriptor) bool { + if len(txo) == 0 { + return false + } else if len(cmpSet) == 0 || len(txo) > len(cmpSet) { + return true + } + + cmpMap := map[string]bool{base64.StdEncoding.EncodeToString(cmpSet[0].Path): true} + + for i := 1; i < len(cmpSet); i++ { + cmpMap[base64.StdEncoding.EncodeToString(cmpSet[i].Path)] = true + } + + for _, v := range txo { + if !cmpMap[base64.StdEncoding.EncodeToString(v.Path)] { + return true + } + } + + return false +} + +type TxnInputOutput struct { + inputs []TxnInput + outputs []TxnOutput +} + +func (io *TxnInputOutput) readSet(txnIdx int) []ReadDescriptor { + return io.inputs[txnIdx] +} + +func (io *TxnInputOutput) writeSet(txnIdx int) []WriteDescriptor { + return io.outputs[txnIdx] +} + +func MakeTxnInputOutput(numTx int) *TxnInputOutput { + return &TxnInputOutput{ + inputs: make([]TxnInput, numTx), + outputs: make([]TxnOutput, numTx), + } +} + +func (io *TxnInputOutput) recordRead(txId int, input []ReadDescriptor) { + io.inputs[txId] = input +} + +func (io *TxnInputOutput) recordWrite(txId int, output []WriteDescriptor) { + io.outputs[txId] = output +} diff --git a/core/state/journal.go b/core/state/journal.go index 57a692dc7f..57393cbcf4 100644 --- a/core/state/journal.go +++ b/core/state/journal.go @@ -143,6 +143,7 @@ type ( func (ch createObjectChange) revert(s *StateDB) { delete(s.stateObjects, *ch.account) delete(s.stateObjectsDirty, *ch.account) + MVWrite(s, ch.account.Bytes()) } func (ch createObjectChange) dirtied() *common.Address { @@ -151,6 +152,7 @@ func (ch createObjectChange) dirtied() *common.Address { func (ch resetObjectChange) revert(s *StateDB) { s.setStateObject(ch.prev) + MVWrite(s, ch.prev.address.Bytes()) if !ch.prevdestruct && s.snap != nil { delete(s.snapDestructs, ch.prev.addrHash) } @@ -165,6 +167,8 @@ func (ch suicideChange) revert(s *StateDB) { if obj != nil { obj.suicided = ch.prev obj.setBalance(ch.prevbalance) + MVWrite(s, subPath(ch.account.Bytes(), suicidePath)) + MVWrite(s, subPath(ch.account.Bytes(), balancePath)) } } @@ -183,6 +187,7 @@ func (ch touchChange) dirtied() *common.Address { func (ch balanceChange) revert(s *StateDB) { s.getStateObject(*ch.account).setBalance(ch.prev) + MVWrite(s, subPath(ch.account.Bytes(), balancePath)) } func (ch balanceChange) dirtied() *common.Address { @@ -191,6 +196,7 @@ func (ch balanceChange) dirtied() *common.Address { func (ch nonceChange) revert(s *StateDB) { s.getStateObject(*ch.account).setNonce(ch.prev) + MVWrite(s, subPath(ch.account.Bytes(), noncePath)) } func (ch nonceChange) dirtied() *common.Address { @@ -199,6 +205,7 @@ func (ch nonceChange) dirtied() *common.Address { func (ch codeChange) revert(s *StateDB) { s.getStateObject(*ch.account).setCode(common.BytesToHash(ch.prevhash), ch.prevcode) + MVWrite(s, subPath(ch.account.Bytes(), codePath)) } func (ch codeChange) dirtied() *common.Address { @@ -207,6 +214,7 @@ func (ch codeChange) dirtied() *common.Address { func (ch storageChange) revert(s *StateDB) { s.getStateObject(*ch.account).setState(ch.key, ch.prevalue) + MVWrite(s, append(ch.account.Bytes(), ch.key.Bytes()...)) } func (ch storageChange) dirtied() *common.Address { diff --git a/core/state/statedb.go b/core/state/statedb.go index c236a79b5a..2d6250994c 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -25,6 +25,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/blockstm" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/state/snapshot" "github.com/ethereum/go-ethereum/core/types" @@ -79,6 +80,12 @@ type StateDB struct { stateObjectsPending map[common.Address]struct{} // State objects finalized but not yet written to the trie stateObjectsDirty map[common.Address]struct{} // State objects modified in the current execution + // Block-stm related fields + mvHashmap *blockstm.MVHashMap + incarnation int + readMap map[string]blockstm.ReadDescriptor + writeMap map[string]blockstm.WriteDescriptor + // DB error. // State objects are used by the consensus core and VM which are // unable to deal with database-level errors. Any error that occurs @@ -154,6 +161,159 @@ func New(root common.Hash, db Database, snaps *snapshot.Tree) (*StateDB, error) return sdb, nil } +func NewWithMVHashmap(root common.Hash, db Database, snaps *snapshot.Tree, mvhm *blockstm.MVHashMap) (*StateDB, error) { + if sdb, err := New(root, db, snaps); err != nil { + return nil, err + } else { + sdb.mvHashmap = mvhm + return sdb, nil + } +} + +func (sdb *StateDB) SetMVHashmap(mvhm *blockstm.MVHashMap) { + sdb.mvHashmap = mvhm +} + +func (s *StateDB) MVWriteList() []blockstm.WriteDescriptor { + writes := make([]blockstm.WriteDescriptor, 0, len(s.writeMap)) + + for _, v := range s.writeMap { + writes = append(writes, v) + } + + return writes +} + +func (s *StateDB) MVReadList() []blockstm.ReadDescriptor { + reads := make([]blockstm.ReadDescriptor, 0, len(s.readMap)) + + for _, v := range s.readMap { + reads = append(reads, v) + } + + return reads +} + +func (s *StateDB) ensureReadMap() { + if s.readMap == nil { + s.readMap = make(map[string]blockstm.ReadDescriptor) + } +} + +func (s *StateDB) ensureWriteMap() { + if s.writeMap == nil { + s.writeMap = make(map[string]blockstm.WriteDescriptor) + } +} + +func MVRead[T any](s *StateDB, k []byte, defaultV T, readStorage func(s *StateDB) T) (v T) { + if s.mvHashmap == nil { + return readStorage(s) + } + + s.ensureReadMap() + + if s.writeMap != nil { + if _, ok := s.writeMap[string(k)]; ok { + return readStorage(s) + } + } + + res := s.mvHashmap.Read(k, s.txIndex) + + var rd blockstm.ReadDescriptor + + rd.V = blockstm.Version{ + TxnIndex: res.DepIdx(), + Incarnation: res.Incarnation(), + } + + rd.Path = k + + switch res.Status() { + case blockstm.MVReadResultDone: + { + v = readStorage(res.Value().(*StateDB)) + rd.Kind = blockstm.ReadKindMap + } + case blockstm.MVReadResultDependency: + { + return defaultV + } + case blockstm.MVReadResultNone: + { + v = readStorage(s) + rd.Kind = blockstm.ReadKindStorage + } + default: + return defaultV + } + + mk := string(k) + // TODO: I assume we don't want to overwrite an existing read because this could - for example - change a storage + // read to map if the same value is read multiple times. + if _, ok := s.readMap[mk]; !ok { + s.readMap[mk] = rd + } + + return +} + +func MVWrite(s *StateDB, k []byte) { + if s.mvHashmap != nil { + s.ensureWriteMap() + s.mvHashmap.Write(k, s.Version(), s) + s.writeMap[string(k)] = blockstm.WriteDescriptor{ + Path: k, + V: s.Version(), + Val: s, + } + } +} + +func MVWritten(s *StateDB, k []byte) bool { + if s.mvHashmap == nil || s.writeMap == nil { + return false + } + + _, ok := s.writeMap[string(k)] + + return ok +} + +func (sw *StateDB) ApplyMVWriteSet(writes []blockstm.WriteDescriptor) { + for i := range writes { + path := writes[i].Path + sr := writes[i].Val.(*StateDB) + + keyLength := len(path) + + if keyLength == common.AddressLength { + sw.GetOrNewStateObject(common.BytesToAddress(path)) + } else if keyLength == (common.AddressLength + common.HashLength) { + addr := common.BytesToAddress(path[:common.AddressLength]) + subPath := common.BytesToHash(path[common.AddressLength:]) + sw.SetState(addr, subPath, sr.GetState(addr, subPath)) + } else { + addr := common.BytesToAddress(path[:common.AddressLength]) + switch path[keyLength-1] { + case balancePath: + sw.SetBalance(addr, sr.GetBalance(addr)) + case noncePath: + sw.SetNonce(addr, sr.GetNonce(addr)) + case codePath: + sw.SetCode(addr, sr.GetCode(addr)) + case suicidePath: + if suicided := sr.HasSuicided(addr); suicided { + sw.Suicide(addr) + } + default: + panic(fmt.Errorf("unknown key type: %d", path[keyLength-1])) + } + } + } +} + // StartPrefetcher initializes a new trie prefetcher to pull in nodes from the // state trie concurrently while the state is mutated so that when we reach the // commit phase, most of the needed data is already hot. @@ -257,22 +417,48 @@ func (s *StateDB) Empty(addr common.Address) bool { return so == nil || so.empty() } +// Create a unique path for special fields (e.g. balance, code) in a state object. +func subPath(prefix []byte, s uint8) []byte { + path := append(prefix, common.Hash{}.Bytes()...) // append a full empty hash to avoid collision with storage state + path = append(path, s) // append the special field identifier + + return path +} + +const balancePath = 1 +const noncePath = 2 +const codePath = 3 +const suicidePath = 4 + // GetBalance retrieves the balance from the given address or 0 if object not found func (s *StateDB) GetBalance(addr common.Address) *big.Int { - stateObject := s.getStateObject(addr) - if stateObject != nil { - return stateObject.Balance() + if s.getStateObject(addr) == nil { + return common.Big0 } - return common.Big0 + + return MVRead(s, subPath(addr.Bytes(), balancePath), common.Big0, func(s *StateDB) *big.Int { + stateObject := s.getStateObject(addr) + if stateObject != nil { + return stateObject.Balance() + } + + return common.Big0 + }) } func (s *StateDB) GetNonce(addr common.Address) uint64 { - stateObject := s.getStateObject(addr) - if stateObject != nil { - return stateObject.Nonce() + if s.getStateObject(addr) == nil { + return 0 } - return 0 + return MVRead(s, subPath(addr.Bytes(), noncePath), 0, func(s *StateDB) uint64 { + stateObject := s.getStateObject(addr) + if stateObject != nil { + return stateObject.Nonce() + } + + return 0 + }) } // TxIndex returns the current transaction index set by Prepare. @@ -280,37 +466,68 @@ func (s *StateDB) TxIndex() int { return s.txIndex } -func (s *StateDB) GetCode(addr common.Address) []byte { - stateObject := s.getStateObject(addr) - if stateObject != nil { - return stateObject.Code(s.db) +func (s *StateDB) Version() blockstm.Version { + return blockstm.Version{ + TxnIndex: s.txIndex, + Incarnation: s.incarnation, } - return nil +} + +func (s *StateDB) GetCode(addr common.Address) []byte { + if s.getStateObject(addr) == nil { + return nil + } + + return MVRead(s, subPath(addr.Bytes(), codePath), nil, func(s *StateDB) []byte { + stateObject := s.getStateObject(addr) + if stateObject != nil { + return stateObject.Code(s.db) + } + return nil + }) } func (s *StateDB) GetCodeSize(addr common.Address) int { - stateObject := s.getStateObject(addr) - if stateObject != nil { - return stateObject.CodeSize(s.db) + if s.getStateObject(addr) == nil { + return 0 } - return 0 + + return MVRead(s, subPath(addr.Bytes(), codePath), 0, func(s *StateDB) int { + stateObject := s.getStateObject(addr) + if stateObject != nil { + return stateObject.CodeSize(s.db) + } + return 0 + }) } func (s *StateDB) GetCodeHash(addr common.Address) common.Hash { - stateObject := s.getStateObject(addr) - if stateObject == nil { + if s.getStateObject(addr) == nil { return common.Hash{} } - return common.BytesToHash(stateObject.CodeHash()) + + return MVRead(s, subPath(addr.Bytes(), codePath), common.Hash{}, func(s *StateDB) common.Hash { + stateObject := s.getStateObject(addr) + if stateObject == nil { + return common.Hash{} + } + return common.BytesToHash(stateObject.CodeHash()) + }) } // GetState retrieves a value from the given account's storage trie. func (s *StateDB) GetState(addr common.Address, hash common.Hash) common.Hash { - stateObject := s.getStateObject(addr) - if stateObject != nil { - return stateObject.GetState(s.db, hash) + if s.getStateObject(addr) == nil { + return common.Hash{} } - return common.Hash{} + + return MVRead(s, append(addr.Bytes(), hash.Bytes()...), common.Hash{}, func(s *StateDB) common.Hash { + stateObject := s.getStateObject(addr) + if stateObject != nil { + return stateObject.GetState(s.db, hash) + } + return common.Hash{} + }) } // GetProof returns the Merkle proof for a given account. @@ -338,11 +555,17 @@ func (s *StateDB) GetStorageProof(a common.Address, key common.Hash) ([][]byte, // GetCommittedState retrieves a value from the given account's committed storage trie. func (s *StateDB) GetCommittedState(addr common.Address, hash common.Hash) common.Hash { - stateObject := s.getStateObject(addr) - if stateObject != nil { - return stateObject.GetCommittedState(s.db, hash) + if s.getStateObject(addr) == nil { + return common.Hash{} } - return common.Hash{} + + return MVRead(s, append(addr.Bytes(), hash.Bytes()...), common.Hash{}, func(s *StateDB) common.Hash { + stateObject := s.getStateObject(addr) + if stateObject != nil { + return stateObject.GetCommittedState(s.db, hash) + } + return common.Hash{} + }) } // Database retrieves the low level database supporting the lower level trie ops. @@ -363,11 +586,17 @@ func (s *StateDB) StorageTrie(addr common.Address) Trie { } func (s *StateDB) HasSuicided(addr common.Address) bool { - stateObject := s.getStateObject(addr) - if stateObject != nil { - return stateObject.suicided + if s.getStateObject(addr) == nil { + return false } - return false + + return MVRead(s, subPath(addr.Bytes(), suicidePath), false, func(s *StateDB) bool { + stateObject := s.getStateObject(addr) + if stateObject != nil { + return stateObject.suicided + } + return false + }) } /* @@ -378,7 +607,9 @@ func (s *StateDB) HasSuicided(addr common.Address) bool { func (s *StateDB) AddBalance(addr common.Address, amount *big.Int) { stateObject := s.GetOrNewStateObject(addr) if stateObject != nil { + stateObject = s.mvRecordWritten(stateObject) stateObject.AddBalance(amount) + MVWrite(s, subPath(addr.Bytes(), balancePath)) } } @@ -386,35 +617,45 @@ func (s *StateDB) AddBalance(addr common.Address, amount *big.Int) { func (s *StateDB) SubBalance(addr common.Address, amount *big.Int) { stateObject := s.GetOrNewStateObject(addr) if stateObject != nil { + stateObject = s.mvRecordWritten(stateObject) stateObject.SubBalance(amount) + MVWrite(s, subPath(addr.Bytes(), balancePath)) } } func (s *StateDB) SetBalance(addr common.Address, amount *big.Int) { stateObject := s.GetOrNewStateObject(addr) if stateObject != nil { + stateObject = s.mvRecordWritten(stateObject) stateObject.SetBalance(amount) + MVWrite(s, subPath(addr.Bytes(), balancePath)) } } func (s *StateDB) SetNonce(addr common.Address, nonce uint64) { stateObject := s.GetOrNewStateObject(addr) if stateObject != nil { + stateObject = s.mvRecordWritten(stateObject) stateObject.SetNonce(nonce) + MVWrite(s, subPath(addr.Bytes(), noncePath)) } } func (s *StateDB) SetCode(addr common.Address, code []byte) { stateObject := s.GetOrNewStateObject(addr) if stateObject != nil { + stateObject = s.mvRecordWritten(stateObject) stateObject.SetCode(crypto.Keccak256Hash(code), code) + MVWrite(s, subPath(addr.Bytes(), codePath)) } } func (s *StateDB) SetState(addr common.Address, key, value common.Hash) { stateObject := s.GetOrNewStateObject(addr) if stateObject != nil { + stateObject = s.mvRecordWritten(stateObject) stateObject.SetState(s.db, key, value) + MVWrite(s, append(addr.Bytes(), key.Bytes()...)) } } @@ -437,6 +678,8 @@ func (s *StateDB) Suicide(addr common.Address) bool { if stateObject == nil { return false } + + stateObject = s.mvRecordWritten(stateObject) s.journal.append(suicideChange{ account: &addr, prev: stateObject.suicided, @@ -445,6 +688,9 @@ func (s *StateDB) Suicide(addr common.Address) bool { stateObject.markSuicided() stateObject.data.Balance = new(big.Int) + MVWrite(s, subPath(addr.Bytes(), suicidePath)) + MVWrite(s, subPath(addr.Bytes(), balancePath)) + return true } @@ -501,60 +747,62 @@ func (s *StateDB) getStateObject(addr common.Address) *stateObject { // flag set. This is needed by the state journal to revert to the correct s- // destructed object instead of wiping all knowledge about the state object. func (s *StateDB) getDeletedStateObject(addr common.Address) *stateObject { - // Prefer live objects if any is available - if obj := s.stateObjects[addr]; obj != nil { - return obj - } - // If no live objects are available, attempt to use snapshots - var data *types.StateAccount - if s.snap != nil { - start := time.Now() - acc, err := s.snap.Account(crypto.HashData(s.hasher, addr.Bytes())) - if metrics.EnabledExpensive { - s.SnapshotAccountReads += time.Since(start) + return MVRead(s, addr.Bytes(), nil, func(s *StateDB) *stateObject { + // Prefer live objects if any is available + if obj := s.stateObjects[addr]; obj != nil { + return obj } - if err == nil { - if acc == nil { + // If no live objects are available, attempt to use snapshots + var data *types.StateAccount + if s.snap != nil { // nolint + start := time.Now() + acc, err := s.snap.Account(crypto.HashData(s.hasher, addr.Bytes())) + if metrics.EnabledExpensive { + s.SnapshotAccountReads += time.Since(start) + } + if err == nil { + if acc == nil { + return nil + } + data = &types.StateAccount{ + Nonce: acc.Nonce, + Balance: acc.Balance, + CodeHash: acc.CodeHash, + Root: common.BytesToHash(acc.Root), + } + if len(data.CodeHash) == 0 { + data.CodeHash = emptyCodeHash + } + if data.Root == (common.Hash{}) { + data.Root = emptyRoot + } + } + } + // If snapshot unavailable or reading from it failed, load from the database + if data == nil { + start := time.Now() + enc, err := s.trie.TryGet(addr.Bytes()) + if metrics.EnabledExpensive { + s.AccountReads += time.Since(start) + } + if err != nil { + s.setError(fmt.Errorf("getDeleteStateObject (%x) error: %v", addr.Bytes(), err)) return nil } - data = &types.StateAccount{ - Nonce: acc.Nonce, - Balance: acc.Balance, - CodeHash: acc.CodeHash, - Root: common.BytesToHash(acc.Root), + if len(enc) == 0 { + return nil } - if len(data.CodeHash) == 0 { - data.CodeHash = emptyCodeHash - } - if data.Root == (common.Hash{}) { - data.Root = emptyRoot + data = new(types.StateAccount) + if err := rlp.DecodeBytes(enc, data); err != nil { + log.Error("Failed to decode state object", "addr", addr, "err", err) + return nil } } - } - // If snapshot unavailable or reading from it failed, load from the database - if data == nil { - start := time.Now() - enc, err := s.trie.TryGet(addr.Bytes()) - if metrics.EnabledExpensive { - s.AccountReads += time.Since(start) - } - if err != nil { - s.setError(fmt.Errorf("getDeleteStateObject (%x) error: %v", addr.Bytes(), err)) - return nil - } - if len(enc) == 0 { - return nil - } - data = new(types.StateAccount) - if err := rlp.DecodeBytes(enc, data); err != nil { - log.Error("Failed to decode state object", "addr", addr, "err", err) - return nil - } - } - // Insert into the live set - obj := newObject(s, addr, *data) - s.setStateObject(obj) - return obj + // Insert into the live set + obj := newObject(s, addr, *data) + s.setStateObject(obj) + return obj + }) } func (s *StateDB) setStateObject(object *stateObject) { @@ -570,6 +818,28 @@ func (s *StateDB) GetOrNewStateObject(addr common.Address) *stateObject { return stateObject } +// mvRecordWritten checks whether a state object is already present in the current MV writeMap. +// If yes, it returns the object directly. +// If not, it clones the object and inserts it into the writeMap before returning it. +func (s *StateDB) mvRecordWritten(object *stateObject) *stateObject { + if s.mvHashmap == nil { + return object + } + + addrPath := object.Address().Bytes() + + if MVWritten(s, addrPath) { + return object + } + + // Deepcopy is needed to ensure that objects are not written by multiple transactions at the same time, because + // the input state object can come from a different transaction. + s.setStateObject(object.deepCopy(s)) + MVWrite(s, addrPath) + + return s.stateObjects[object.Address()] +} + // createObject creates a new state object. If there is an existing account with // the given address, it is overwritten and returned as the second return value. func (s *StateDB) createObject(addr common.Address) (newobj, prev *stateObject) { @@ -589,6 +859,7 @@ func (s *StateDB) createObject(addr common.Address) (newobj, prev *stateObject) s.journal.append(resetObjectChange{prev: prev, prevdestruct: prevdestruct}) } s.setStateObject(newobj) + MVWrite(s, addr.Bytes()) if prev != nil && !prev.deleted { return newobj, prev } @@ -609,6 +880,7 @@ func (s *StateDB) CreateAccount(addr common.Address) { newObj, prev := s.createObject(addr) if prev != nil { newObj.setBalance(prev.data.Balance) + MVWrite(s, subPath(addr.Bytes(), balancePath)) } } @@ -738,6 +1010,10 @@ func (s *StateDB) Copy() *StateDB { state.snapStorage[k] = temp } } + + if s.mvHashmap != nil { + state.mvHashmap = s.mvHashmap + } return state } diff --git a/core/state/statedb_test.go b/core/state/statedb_test.go index e9576d4dc4..c1f9f04812 100644 --- a/core/state/statedb_test.go +++ b/core/state/statedb_test.go @@ -29,7 +29,10 @@ import ( "testing" "testing/quick" + "github.com/stretchr/testify/assert" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/blockstm" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" ) @@ -488,6 +491,427 @@ func TestTouchDelete(t *testing.T) { } } +func TestMVHashMapReadWriteDelete(t *testing.T) { + t.Parallel() + + db := NewDatabase(rawdb.NewMemoryDatabase()) + mvhm := blockstm.MakeMVHashMap() + s, _ := NewWithMVHashmap(common.Hash{}, db, nil, mvhm) + + states := []*StateDB{s} + + // Create copies of the original state for each transition + for i := 1; i <= 4; i++ { + sCopy := s.Copy() + sCopy.txIndex = i + states = append(states, sCopy) + } + + addr := common.HexToAddress("0x01") + key := common.HexToHash("0x01") + val := common.HexToHash("0x01") + balance := new(big.Int).SetUint64(uint64(100)) + + // Tx0 read + v := states[0].GetState(addr, key) + + assert.Equal(t, common.Hash{}, v) + + // Tx1 write + states[1].GetOrNewStateObject(addr) + states[1].SetState(addr, key, val) + states[1].SetBalance(addr, balance) + + // Tx1 read + v = states[2].GetState(addr, key) + b := states[2].GetBalance(addr) + + assert.Equal(t, val, v) + assert.Equal(t, balance, b) + + // Tx2 read + v = states[2].GetState(addr, key) + b = states[2].GetBalance(addr) + + assert.Equal(t, val, v) + assert.Equal(t, balance, b) + + // Tx3 delete + states[3].Suicide(addr) + + // Within Tx 3, the state should not change before finalize + v = states[3].GetState(addr, key) + assert.Equal(t, val, v) + + // After finalizing Tx 3, the state will change + states[3].Finalise(false) + v = states[3].GetState(addr, key) + assert.Equal(t, common.Hash{}, v) + + // Tx4 read + v = states[4].GetState(addr, key) + b = states[4].GetBalance(addr) + + assert.Equal(t, common.Hash{}, v) + assert.Equal(t, common.Big0, b) +} + +func TestMVHashMapRevert(t *testing.T) { + t.Parallel() + + db := NewDatabase(rawdb.NewMemoryDatabase()) + mvhm := blockstm.MakeMVHashMap() + s, _ := NewWithMVHashmap(common.Hash{}, db, nil, mvhm) + + states := []*StateDB{s} + + // Create copies of the original state for each transition + for i := 1; i <= 4; i++ { + sCopy := s.Copy() + sCopy.txIndex = i + states = append(states, sCopy) + } + + addr := common.HexToAddress("0x01") + key := common.HexToHash("0x01") + val := common.HexToHash("0x01") + balance := new(big.Int).SetUint64(uint64(100)) + + // Tx0 write + states[0].GetOrNewStateObject(addr) + states[0].SetState(addr, key, val) + states[0].SetBalance(addr, balance) + + // Tx1 perform some ops and then revert + snapshot := states[1].Snapshot() + states[1].AddBalance(addr, new(big.Int).SetUint64(uint64(100))) + states[1].SetState(addr, key, common.HexToHash("0x02")) + v := states[1].GetState(addr, key) + b := states[1].GetBalance(addr) + assert.Equal(t, new(big.Int).SetUint64(uint64(200)), b) + assert.Equal(t, common.HexToHash("0x02"), v) + + states[1].Suicide(addr) + + states[1].RevertToSnapshot(snapshot) + + v = states[1].GetState(addr, key) + b = states[1].GetBalance(addr) + + assert.Equal(t, val, v) + assert.Equal(t, balance, b) + states[1].Finalise(false) + + // Tx2 check the state and balance + v = states[2].GetState(addr, key) + b = states[2].GetBalance(addr) + + assert.Equal(t, val, v) + assert.Equal(t, balance, b) +} + +func TestMVHashMapMarkEstimate(t *testing.T) { + t.Parallel() + + db := NewDatabase(rawdb.NewMemoryDatabase()) + mvhm := blockstm.MakeMVHashMap() + s, _ := NewWithMVHashmap(common.Hash{}, db, nil, mvhm) + + states := []*StateDB{s} + + // Create copies of the original state for each transition + for i := 1; i <= 4; i++ { + sCopy := s.Copy() + sCopy.txIndex = i + states = append(states, sCopy) + } + + addr := common.HexToAddress("0x01") + key := common.HexToHash("0x01") + val := common.HexToHash("0x01") + balance := new(big.Int).SetUint64(uint64(100)) + + // Tx0 read + v := states[0].GetState(addr, key) + assert.Equal(t, common.Hash{}, v) + + // Tx0 write + states[0].SetState(addr, key, val) + v = states[0].GetState(addr, key) + assert.Equal(t, val, v) + + // Tx1 write + states[1].GetOrNewStateObject(addr) + states[1].SetState(addr, key, val) + states[1].SetBalance(addr, balance) + + // Tx2 read + v = states[2].GetState(addr, key) + b := states[2].GetBalance(addr) + + assert.Equal(t, val, v) + assert.Equal(t, balance, b) + + // Tx1 mark estimate + for _, v := range states[1].writeMap { + mvhm.MarkEstimate(v.Path, 1) + } + + // Tx2 read again should get default (empty) vals because its dependency Tx1 is marked as estimate + v = states[2].GetState(addr, key) + b = states[2].GetBalance(addr) + + assert.Equal(t, common.Hash{}, v) + assert.Equal(t, common.Big0, b) + + // Tx1 read again should get Tx0 vals + v = states[1].GetState(addr, key) + assert.Equal(t, val, v) +} + +func TestMVHashMapOverwrite(t *testing.T) { + t.Parallel() + + db := NewDatabase(rawdb.NewMemoryDatabase()) + mvhm := blockstm.MakeMVHashMap() + s, _ := NewWithMVHashmap(common.Hash{}, db, nil, mvhm) + + states := []*StateDB{s} + + // Create copies of the original state for each transition + for i := 1; i <= 4; i++ { + sCopy := s.Copy() + sCopy.txIndex = i + states = append(states, sCopy) + } + + addr := common.HexToAddress("0x01") + key := common.HexToHash("0x01") + val1 := common.HexToHash("0x01") + balance1 := new(big.Int).SetUint64(uint64(100)) + val2 := common.HexToHash("0x02") + balance2 := new(big.Int).SetUint64(uint64(200)) + + // Tx0 write + states[0].GetOrNewStateObject(addr) + states[0].SetState(addr, key, val1) + states[0].SetBalance(addr, balance1) + + // Tx1 write + states[1].SetState(addr, key, val2) + states[1].SetBalance(addr, balance2) + v := states[1].GetState(addr, key) + b := states[1].GetBalance(addr) + + assert.Equal(t, val2, v) + assert.Equal(t, balance2, b) + + // Tx2 read should get Tx1's value + v = states[2].GetState(addr, key) + b = states[2].GetBalance(addr) + + assert.Equal(t, val2, v) + assert.Equal(t, balance2, b) + + // Tx1 delete + for _, v := range states[1].writeMap { + mvhm.Delete(v.Path, 1) + + states[1].writeMap = nil + } + + // Tx2 read should get Tx0's value + v = states[2].GetState(addr, key) + b = states[2].GetBalance(addr) + + assert.Equal(t, val1, v) + assert.Equal(t, balance1, b) + + // Tx1 read should get Tx0's value + v = states[1].GetState(addr, key) + b = states[1].GetBalance(addr) + + assert.Equal(t, val1, v) + assert.Equal(t, balance1, b) + + // Tx0 delete + for _, v := range states[0].writeMap { + mvhm.Delete(v.Path, 0) + + states[0].writeMap = nil + } + + // Tx2 read again should get default vals + v = states[2].GetState(addr, key) + b = states[2].GetBalance(addr) + + assert.Equal(t, common.Hash{}, v) + assert.Equal(t, common.Big0, b) +} + +func TestMVHashMapWriteNoConflict(t *testing.T) { + t.Parallel() + + db := NewDatabase(rawdb.NewMemoryDatabase()) + mvhm := blockstm.MakeMVHashMap() + s, _ := NewWithMVHashmap(common.Hash{}, db, nil, mvhm) + + states := []*StateDB{s} + + // Create copies of the original state for each transition + for i := 1; i <= 4; i++ { + sCopy := s.Copy() + sCopy.txIndex = i + states = append(states, sCopy) + } + + addr := common.HexToAddress("0x01") + key1 := common.HexToHash("0x01") + key2 := common.HexToHash("0x02") + val1 := common.HexToHash("0x01") + balance1 := new(big.Int).SetUint64(uint64(100)) + val2 := common.HexToHash("0x02") + + // Tx0 write + states[0].GetOrNewStateObject(addr) + + // Tx2 write + states[2].SetState(addr, key2, val2) + + // Tx1 write + tx1Snapshot := states[1].Snapshot() + states[1].SetState(addr, key1, val1) + states[1].SetBalance(addr, balance1) + + // Tx1 read + assert.Equal(t, val1, states[1].GetState(addr, key1)) + assert.Equal(t, balance1, states[1].GetBalance(addr)) + // Tx1 should see empty value in key2 + assert.Equal(t, common.Hash{}, states[1].GetState(addr, key2)) + + // Tx2 read + assert.Equal(t, val2, states[2].GetState(addr, key2)) + // Tx2 should see values written by Tx1 + assert.Equal(t, val1, states[2].GetState(addr, key1)) + assert.Equal(t, balance1, states[2].GetBalance(addr)) + + // Tx3 read + assert.Equal(t, val1, states[3].GetState(addr, key1)) + assert.Equal(t, val2, states[3].GetState(addr, key2)) + assert.Equal(t, balance1, states[3].GetBalance(addr)) + + // Tx2 delete + for _, v := range states[2].writeMap { + mvhm.Delete(v.Path, 2) + + states[2].writeMap = nil + } + + assert.Equal(t, val1, states[3].GetState(addr, key1)) + assert.Equal(t, balance1, states[3].GetBalance(addr)) + assert.Equal(t, common.Hash{}, states[3].GetState(addr, key2)) + + // Tx1 revert + states[1].RevertToSnapshot(tx1Snapshot) + + assert.Equal(t, common.Hash{}, states[3].GetState(addr, key1)) + assert.Equal(t, common.Hash{}, states[3].GetState(addr, key2)) + assert.Equal(t, common.Big0, states[3].GetBalance(addr)) + + // Tx1 delete + for _, v := range states[1].writeMap { + mvhm.Delete(v.Path, 1) + + states[1].writeMap = nil + } + + assert.Equal(t, common.Hash{}, states[3].GetState(addr, key1)) + assert.Equal(t, common.Hash{}, states[3].GetState(addr, key2)) + assert.Equal(t, common.Big0, states[3].GetBalance(addr)) +} + +func TestApplyMVWriteSet(t *testing.T) { + t.Parallel() + + db := NewDatabase(rawdb.NewMemoryDatabase()) + mvhm := blockstm.MakeMVHashMap() + s, _ := NewWithMVHashmap(common.Hash{}, db, nil, mvhm) + + sClean := s.Copy() + sClean.mvHashmap = nil + + sSingleProcess := sClean.Copy() + + states := []*StateDB{s} + + // Create copies of the original state for each transition + for i := 1; i <= 4; i++ { + sCopy := s.Copy() + sCopy.txIndex = i + states = append(states, sCopy) + } + + addr1 := common.HexToAddress("0x01") + addr2 := common.HexToAddress("0x02") + key1 := common.HexToHash("0x01") + key2 := common.HexToHash("0x02") + val1 := common.HexToHash("0x01") + balance1 := new(big.Int).SetUint64(uint64(100)) + val2 := common.HexToHash("0x02") + balance2 := new(big.Int).SetUint64(uint64(200)) + code := []byte{1, 2, 3} + + // Tx0 write + states[0].SetState(addr1, key1, val1) + states[0].SetBalance(addr1, balance1) + states[0].SetState(addr2, key2, val2) + + sSingleProcess.SetState(addr1, key1, val1) + sSingleProcess.SetBalance(addr1, balance1) + sSingleProcess.SetState(addr2, key2, val2) + + sClean.ApplyMVWriteSet(states[0].MVWriteList()) + + assert.Equal(t, sSingleProcess.IntermediateRoot(false), sClean.IntermediateRoot(false)) + + // Tx1 write + states[1].SetState(addr1, key2, val2) + states[1].SetBalance(addr1, balance2) + states[1].SetNonce(addr1, 1) + + sSingleProcess.SetState(addr1, key2, val2) + sSingleProcess.SetBalance(addr1, balance2) + sSingleProcess.SetNonce(addr1, 1) + + sClean.ApplyMVWriteSet(states[1].MVWriteList()) + + assert.Equal(t, sSingleProcess.IntermediateRoot(false), sClean.IntermediateRoot(false)) + + // Tx2 write + states[2].SetState(addr1, key1, val2) + states[2].SetBalance(addr1, balance2) + states[2].SetNonce(addr1, 2) + + sSingleProcess.SetState(addr1, key1, val2) + sSingleProcess.SetBalance(addr1, balance2) + sSingleProcess.SetNonce(addr1, 2) + + sClean.ApplyMVWriteSet(states[2].MVWriteList()) + + assert.Equal(t, sSingleProcess.IntermediateRoot(false), sClean.IntermediateRoot(false)) + + // Tx3 write + states[3].Suicide(addr2) + states[3].SetCode(addr1, code) + + sSingleProcess.Suicide(addr2) + sSingleProcess.SetCode(addr1, code) + + sClean.ApplyMVWriteSet(states[3].MVWriteList()) + + assert.Equal(t, sSingleProcess.IntermediateRoot(false), sClean.IntermediateRoot(false)) +} + // TestCopyOfCopy tests that modified objects are carried over to the copy, and the copy of the copy. // See https://github.com/ethereum/go-ethereum/pull/15225#issuecomment-380191512 func TestCopyOfCopy(t *testing.T) { diff --git a/go.mod b/go.mod index 7a643a251c..fa21583ce2 100644 --- a/go.mod +++ b/go.mod @@ -105,6 +105,8 @@ require ( github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect github.com/deepmap/oapi-codegen v1.8.2 // indirect github.com/dlclark/regexp2 v1.4.1-0.20201116162257-a2a8dda75c91 // indirect + github.com/emirpasic/gods v1.18.1 + github.com/go-kit/kit v0.9.0 // indirect github.com/go-logfmt/logfmt v0.5.0 // indirect github.com/go-ole/go-ole v1.2.1 // indirect github.com/go-sourcemap/sourcemap v2.1.3+incompatible // indirect diff --git a/go.sum b/go.sum index dc419821c6..6d28e061ef 100644 --- a/go.sum +++ b/go.sum @@ -144,6 +144,8 @@ github.com/dop251/goja_nodejs v0.0.0-20210225215109-d91c329300e7/go.mod h1:hn7BA github.com/eclipse/paho.mqtt.golang v1.2.0/go.mod h1:H9keYFcgq3Qr5OUJm/JZI/i6U7joQ8SYLhZwfeOo6Ts= github.com/edsrzf/mmap-go v1.0.0 h1:CEBF7HpRnUCSJgGUb5h1Gm7e3VkmVDrR8lvWVLtrOFw= github.com/edsrzf/mmap-go v1.0.0/go.mod h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaBNrHW5M= +github.com/emirpasic/gods v1.18.1 h1:FXtiHYKDGKCW2KzwZKx0iC0PQmdlorYgdFG9jPXJ1Bc= +github.com/emirpasic/gods v1.18.1/go.mod h1:8tpGGwCnJ5H4r6BWwaV6OrWmMoPhUl5jm/FMNAnJvWQ= github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=