forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-17 10:20:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

core, trie: port snap sync changes #24898 (#1096)

Browse source 
core, eth, les, trie: rework snap sync

Co-authored-by: rjl493456442 <garyrong0905@gmail.com>
This commit is contained in:
Daniel Liu 2025-08-26 15:20:39 +08:00 committed by GitHub
parent 3d69b27ae4
commit 18d2dbad89
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 840 additions and 483 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
XDCx/tradingstate/state_liquidationprice.go
View file

@ -136,7 +136,7 @@ func (l *liquidationPriceState) updateRoot(db Database) error {
if l.dbErr != nil { if l.dbErr != nil {
return l.dbErr return l.dbErr
} }
root, err := l.trie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash) error { root, err := l.trie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash, _ []byte) error {
var orderList orderList var orderList orderList
if err := rlp.DecodeBytes(leaf, &orderList); err != nil { if err := rlp.DecodeBytes(leaf, &orderList); err != nil {
return nil return nil

6
XDCx/tradingstate/state_orderbook.go
View file

@ -245,7 +245,7 @@ func (te *tradingExchanges) CommitAsksTrie(db Database) error {
if te.dbErr != nil { if te.dbErr != nil {
return te.dbErr return te.dbErr
} }
root, err := te.asksTrie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash) error { root, err := te.asksTrie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash, _ []byte) error {
var orderList orderList var orderList orderList
if err := rlp.DecodeBytes(leaf, &orderList); err != nil { if err := rlp.DecodeBytes(leaf, &orderList); err != nil {
return nil return nil
@ -307,7 +307,7 @@ func (te *tradingExchanges) CommitBidsTrie(db Database) error {
if te.dbErr != nil { if te.dbErr != nil {
return te.dbErr return te.dbErr
} }
root, err := te.bidsTrie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash) error { root, err := te.bidsTrie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash, _ []byte) error {
var orderList orderList var orderList orderList
if err := rlp.DecodeBytes(leaf, &orderList); err != nil { if err := rlp.DecodeBytes(leaf, &orderList); err != nil {
return nil return nil
@ -783,7 +783,7 @@ func (t *tradingExchanges) CommitLiquidationPriceTrie(db Database) error {
if t.dbErr != nil { if t.dbErr != nil {
return t.dbErr return t.dbErr
} }
root, err := t.liquidationPriceTrie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash) error { root, err := t.liquidationPriceTrie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash, _ []byte) error {
var orderList orderList var orderList orderList
if err := rlp.DecodeBytes(leaf, &orderList); err != nil { if err := rlp.DecodeBytes(leaf, &orderList); err != nil {
return nil return nil

2
XDCx/tradingstate/statedb.go
View file

@ -589,7 +589,7 @@ func (t *TradingStateDB) Commit() (root common.Hash, err error) {
} }
} }
// Write trie changes. // Write trie changes.
root, err = t.trie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash) error { root, err = t.trie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash, _ []byte) error {
var exchange tradingExchangeObject var exchange tradingExchangeObject
if err := rlp.DecodeBytes(leaf, &exchange); err != nil { if err := rlp.DecodeBytes(leaf, &exchange); err != nil {
return nil return nil

6
XDCxlending/lendingstate/state_lendingbook.go
View file

@ -472,7 +472,7 @@ func (le *lendingExchangeState) CommitInvestingTrie(db Database) error {
if le.dbErr != nil { if le.dbErr != nil {
return le.dbErr return le.dbErr
} }
root, err := le.investingTrie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash) error { root, err := le.investingTrie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash, _ []byte) error {
var orderList itemList var orderList itemList
if err := rlp.DecodeBytes(leaf, &orderList); err != nil { if err := rlp.DecodeBytes(leaf, &orderList); err != nil {
return nil return nil
@ -493,7 +493,7 @@ func (le *lendingExchangeState) CommitBorrowingTrie(db Database) error {
if le.dbErr != nil { if le.dbErr != nil {
return le.dbErr return le.dbErr
} }
root, err := le.borrowingTrie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash) error { root, err := le.borrowingTrie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash, _ []byte) error {
var orderList itemList var orderList itemList
if err := rlp.DecodeBytes(leaf, &orderList); err != nil { if err := rlp.DecodeBytes(leaf, &orderList); err != nil {
return nil return nil
@ -514,7 +514,7 @@ func (le *lendingExchangeState) CommitLiquidationTimeTrie(db Database) error {
if le.dbErr != nil { if le.dbErr != nil {
return le.dbErr return le.dbErr
} }
root, err := le.liquidationTimeTrie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash) error { root, err := le.liquidationTimeTrie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash, _ []byte) error {
var orderList itemList var orderList itemList
if err := rlp.DecodeBytes(leaf, &orderList); err != nil { if err := rlp.DecodeBytes(leaf, &orderList); err != nil {
return nil return nil

2
XDCxlending/lendingstate/statedb.go
View file

@ -578,7 +578,7 @@ func (ls *LendingStateDB) Commit() (root common.Hash, err error) {
} }
} }
// Write trie changes. // Write trie changes.
root, err = ls.trie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash) error { root, err = ls.trie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash, _ []byte) error {
var exchange lendingObject var exchange lendingObject
if err := rlp.DecodeBytes(leaf, &exchange); err != nil { if err := rlp.DecodeBytes(leaf, &exchange); err != nil {
return nil return nil

2
core/state/statedb.go
View file

@ -835,7 +835,7 @@ func (s *StateDB) Commit(deleteEmptyObjects bool) (common.Hash, error) {
// Write the account trie changes, measuing the amount of wasted time // Write the account trie changes, measuing the amount of wasted time
start := time.Now() start := time.Now()
root, accountCommitted, err := s.trie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash) error { root, accountCommitted, err := s.trie.Commit(func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash, _ []byte) error {
var account types.StateAccount var account types.StateAccount
if err := rlp.DecodeBytes(leaf, &account); err != nil { if err := rlp.DecodeBytes(leaf, &account); err != nil {
return nil return nil

16
core/state/sync.go
View file

@ -27,20 +27,20 @@ import (
) )
// NewStateSync create a new state trie download scheduler. // NewStateSync create a new state trie download scheduler.
func NewStateSync(root common.Hash, database ethdb.KeyValueReader, onLeaf func(paths [][]byte, leaf []byte) error) *trie.Sync { func NewStateSync(root common.Hash, database ethdb.KeyValueReader, onLeaf func(keys [][]byte, leaf []byte) error) *trie.Sync {
// Register the storage slot callback if the external callback is specified. // Register the storage slot callback if the external callback is specified.
var onSlot func(paths [][]byte, hexpath []byte, leaf []byte, parent common.Hash) error var onSlot func(keys [][]byte, path []byte, leaf []byte, parent common.Hash, parentPath []byte) error
if onLeaf != nil { if onLeaf != nil {
onSlot = func(paths [][]byte, hexpath []byte, leaf []byte, parent common.Hash) error { onSlot = func(keys [][]byte, path []byte, leaf []byte, parent common.Hash, parentPath []byte) error {
return onLeaf(paths, leaf) return onLeaf(keys, leaf)
} }
} }
// Register the account callback to connect the state trie and the storage // Register the account callback to connect the state trie and the storage
// trie belongs to the contract. // trie belongs to the contract.
var syncer *trie.Sync var syncer *trie.Sync
onAccount := func(paths [][]byte, hexpath []byte, leaf []byte, parent common.Hash) error { onAccount := func(keys [][]byte, path []byte, leaf []byte, parent common.Hash, parentPath []byte) error {
if onLeaf != nil { if onLeaf != nil {
if err := onLeaf(paths, leaf); err != nil { if err := onLeaf(keys, leaf); err != nil {
return err return err
} }
} }
@ -48,8 +48,8 @@ func NewStateSync(root common.Hash, database ethdb.KeyValueReader, onLeaf func(p
if err := rlp.Decode(bytes.NewReader(leaf), &obj); err != nil { if err := rlp.Decode(bytes.NewReader(leaf), &obj); err != nil {
return err return err
} }
syncer.AddSubTrie(obj.Root, hexpath, parent, onSlot) syncer.AddSubTrie(obj.Root, path, parent, parentPath, onSlot)
syncer.AddCodeEntry(common.BytesToHash(obj.CodeHash), hexpath, parent) syncer.AddCodeEntry(common.BytesToHash(obj.CodeHash), path, parent, parentPath)
return nil return nil
} }
syncer = trie.NewSync(root, database, onAccount) syncer = trie.NewSync(root, database, onAccount)

496
core/state/sync_test.go
View file

@ -133,8 +133,8 @@ func checkStateConsistency(db ethdb.Database, root common.Hash) error {
// Tests that an empty state is not scheduled for syncing. // Tests that an empty state is not scheduled for syncing.
func TestEmptyStateSync(t *testing.T) { func TestEmptyStateSync(t *testing.T) {
sync := NewStateSync(types.EmptyRootHash, rawdb.NewMemoryDatabase(), nil) sync := NewStateSync(types.EmptyRootHash, rawdb.NewMemoryDatabase(), nil)
if nodes, paths, codes := sync.Missing(1); len(nodes) != 0 || len(paths) != 0 || len(codes) != 0 { if paths, nodes, codes := sync.Missing(1); len(paths) != 0 || len(nodes) != 0 || len(codes) != 0 {
t.Errorf(" content requested for empty state: %v, %v, %v", nodes, paths, codes) t.Errorf("content requested for empty state: %v, %v, %v", nodes, paths, codes)
} }
} }
@ -159,6 +159,14 @@ func TestIterativeStateSyncBatchedByPath(t *testing.T) {
testIterativeStateSync(t, 100, false, true) testIterativeStateSync(t, 100, false, true)
} }
// stateElement represents the element in the state trie(bytecode or trie node).
type stateElement struct {
path string
hash common.Hash
code common.Hash
syncPath trie.SyncPath
}
func testIterativeStateSync(t *testing.T, count int, commit bool, bypath bool) { func testIterativeStateSync(t *testing.T, count int, commit bool, bypath bool) {
// Create a random state to copy // Create a random state to copy
srcDb, srcRoot, srcAccounts := makeTestState() srcDb, srcRoot, srcAccounts := makeTestState()
@ -171,54 +179,73 @@ func testIterativeStateSync(t *testing.T, count int, commit bool, bypath bool) {
dstDb := rawdb.NewMemoryDatabase() dstDb := rawdb.NewMemoryDatabase()
sched := NewStateSync(srcRoot, dstDb, nil) sched := NewStateSync(srcRoot, dstDb, nil)
nodes, paths, codes := sched.Missing(count)
var ( var (
hashQueue []common.Hash nodeElements []stateElement
pathQueue []trie.SyncPath codeElements []stateElement
) )
if !bypath { paths, nodes, codes := sched.Missing(count)
hashQueue = append(append(hashQueue[:0], nodes...), codes...) for i := 0; i < len(paths); i++ {
} else { nodeElements = append(nodeElements, stateElement{
hashQueue = append(hashQueue[:0], codes...) path: paths[i],
pathQueue = append(pathQueue[:0], paths...) hash: nodes[i],
syncPath: trie.NewSyncPath([]byte(paths[i])),
})
} }
for len(hashQueue)+len(pathQueue) > 0 { for i := 0; i < len(codes); i++ {
results := make([]trie.SyncResult, len(hashQueue)+len(pathQueue)) codeElements = append(codeElements, stateElement{
for i, hash := range hashQueue { code: codes[i],
data, err := srcDb.TrieDB().Node(hash) })
}
for len(nodeElements)+len(codeElements) > 0 {
var (
nodeResults = make([]trie.NodeSyncResult, len(nodeElements))
codeResults = make([]trie.CodeSyncResult, len(codeElements))
)
for i, element := range codeElements {
data, err := srcDb.ContractCode(common.Hash{}, element.code)
if err != nil { if err != nil {
data, err = srcDb.ContractCode(common.Hash{}, hash) t.Fatalf("failed to retrieve contract bytecode for hash %x", element.code)
} }
if err != nil { codeResults[i] = trie.CodeSyncResult{Hash: element.code, Data: data}
t.Fatalf("failed to retrieve node data for hash %x", hash)
}
results[i] = trie.SyncResult{Hash: hash, Data: data}
} }
for i, path := range pathQueue { for i, node := range nodeElements {
if len(path) == 1 { if bypath {
data, _, err := srcTrie.TryGetNode(path[0]) if len(node.syncPath) == 1 {
if err != nil { data, _, err := srcTrie.TryGetNode(node.syncPath[0])
t.Fatalf("failed to retrieve node data for path %x: %v", path, err) if err != nil {
t.Fatalf("failed to retrieve node data for path %x: %v", node.syncPath[0], err)
}
nodeResults[i] = trie.NodeSyncResult{Path: node.path, Data: data}
} else {
var acc types.StateAccount
if err := rlp.DecodeBytes(srcTrie.Get(node.syncPath[0]), &acc); err != nil {
t.Fatalf("failed to decode account on path %x: %v", node.syncPath[0], err)
}
stTrie, err := trie.New(common.BytesToHash(node.syncPath[0]), acc.Root, srcDb.TrieDB())
if err != nil {
t.Fatalf("failed to retriev storage trie for path %x: %v", node.syncPath[1], err)
}
data, _, err := stTrie.TryGetNode(node.syncPath[1])
if err != nil {
t.Fatalf("failed to retrieve node data for path %x: %v", node.syncPath[1], err)
}
nodeResults[i] = trie.NodeSyncResult{Path: node.path, Data: data}
} }
results[len(hashQueue)+i] = trie.SyncResult{Hash: crypto.Keccak256Hash(data), Data: data}
} else { } else {
var acc types.StateAccount data, err := srcDb.TrieDB().Node(node.hash)
if err := rlp.DecodeBytes(srcTrie.Get(path[0]), &acc); err != nil {
t.Fatalf("failed to decode account on path %x: %v", path, err)
}
stTrie, err := trie.New(common.BytesToHash(path[0]), acc.Root, srcDb.TrieDB())
if err != nil { if err != nil {
t.Fatalf("failed to retriev storage trie for path %x: %v", path, err) t.Fatalf("failed to retrieve node data for key %v", []byte(node.path))
} }
data, _, err := stTrie.TryGetNode(path[1]) nodeResults[i] = trie.NodeSyncResult{Path: node.path, Data: data}
if err != nil {
t.Fatalf("failed to retrieve node data for path %x: %v", path, err)
}
results[len(hashQueue)+i] = trie.SyncResult{Hash: crypto.Keccak256Hash(data), Data: data}
} }
} }
for _, result := range results { for _, result := range codeResults {
if err := sched.Process(result); err != nil { if err := sched.ProcessCode(result); err != nil {
t.Errorf("failed to process result %v", err)
}
}
for _, result := range nodeResults {
if err := sched.ProcessNode(result); err != nil {
t.Errorf("failed to process result %v", err) t.Errorf("failed to process result %v", err)
} }
} }
@ -228,12 +255,20 @@ func testIterativeStateSync(t *testing.T, count int, commit bool, bypath bool) {
} }
batch.Write() batch.Write()
nodes, paths, codes = sched.Missing(count) paths, nodes, codes = sched.Missing(count)
if !bypath { nodeElements = nodeElements[:0]
hashQueue = append(append(hashQueue[:0], nodes...), codes...) for i := 0; i < len(paths); i++ {
} else { nodeElements = append(nodeElements, stateElement{
hashQueue = append(hashQueue[:0], codes...) path: paths[i],
pathQueue = append(pathQueue[:0], paths...) hash: nodes[i],
syncPath: trie.NewSyncPath([]byte(paths[i])),
})
}
codeElements = codeElements[:0]
for i := 0; i < len(codes); i++ {
codeElements = append(codeElements, stateElement{
code: codes[i],
})
} }
} }
// Cross check that the two states are in sync // Cross check that the two states are in sync
@ -250,26 +285,58 @@ func TestIterativeDelayedStateSync(t *testing.T) {
dstDb := rawdb.NewMemoryDatabase() dstDb := rawdb.NewMemoryDatabase()
sched := NewStateSync(srcRoot, dstDb, nil) sched := NewStateSync(srcRoot, dstDb, nil)
nodes, _, codes := sched.Missing(0) var (
queue := append(append([]common.Hash{}, nodes...), codes...) nodeElements []stateElement
codeElements []stateElement
for len(queue) > 0 { )
paths, nodes, codes := sched.Missing(0)
for i := 0; i < len(paths); i++ {
nodeElements = append(nodeElements, stateElement{
path: paths[i],
hash: nodes[i],
syncPath: trie.NewSyncPath([]byte(paths[i])),
})
}
for i := 0; i < len(codes); i++ {
codeElements = append(codeElements, stateElement{
code: codes[i],
})
}
for len(nodeElements)+len(codeElements) > 0 {
// Sync only half of the scheduled nodes // Sync only half of the scheduled nodes
results := make([]trie.SyncResult, len(queue)/2+1) var nodeProcessd int
for i, hash := range queue[:len(results)] { var codeProcessd int
data, err := srcDb.TrieDB().Node(hash) if len(codeElements) > 0 {
if err != nil { codeResults := make([]trie.CodeSyncResult, len(codeElements)/2+1)
data, err = srcDb.ContractCode(common.Hash{}, hash) for i, element := range codeElements[:len(codeResults)] {
data, err := srcDb.ContractCode(common.Hash{}, element.code)
if err != nil {
t.Fatalf("failed to retrieve contract bytecode for %x", element.code)
}
codeResults[i] = trie.CodeSyncResult{Hash: element.code, Data: data}
} }
if err != nil { for _, result := range codeResults {
t.Fatalf("failed to retrieve node data for %x", hash) if err := sched.ProcessCode(result); err != nil {
t.Fatalf("failed to process result %v", err)
}
} }
results[i] = trie.SyncResult{Hash: hash, Data: data} codeProcessd = len(codeResults)
} }
for _, result := range results { if len(nodeElements) > 0 {
if err := sched.Process(result); err != nil { nodeResults := make([]trie.NodeSyncResult, len(nodeElements)/2+1)
t.Fatalf("failed to process result %v", err) for i, element := range nodeElements[:len(nodeResults)] {
data, err := srcDb.TrieDB().Node(element.hash)
if err != nil {
t.Fatalf("failed to retrieve contract bytecode for %x", element.code)
}
nodeResults[i] = trie.NodeSyncResult{Path: element.path, Data: data}
} }
for _, result := range nodeResults {
if err := sched.ProcessNode(result); err != nil {
t.Fatalf("failed to process result %v", err)
}
}
nodeProcessd = len(nodeResults)
} }
batch := dstDb.NewBatch() batch := dstDb.NewBatch()
if err := sched.Commit(batch); err != nil { if err := sched.Commit(batch); err != nil {
@ -277,8 +344,21 @@ func TestIterativeDelayedStateSync(t *testing.T) {
} }
batch.Write() batch.Write()
nodes, _, codes = sched.Missing(0) paths, nodes, codes = sched.Missing(0)
queue = append(append(queue[len(results):], nodes...), codes...) nodeElements = nodeElements[nodeProcessd:]
for i := 0; i < len(paths); i++ {
nodeElements = append(nodeElements, stateElement{
path: paths[i],
hash: nodes[i],
syncPath: trie.NewSyncPath([]byte(paths[i])),
})
}
codeElements = codeElements[codeProcessd:]
for i := 0; i < len(codes); i++ {
codeElements = append(codeElements, stateElement{
code: codes[i],
})
}
} }
// Cross check that the two states are in sync // Cross check that the two states are in sync
checkStateAccounts(t, dstDb, srcRoot, srcAccounts) checkStateAccounts(t, dstDb, srcRoot, srcAccounts)
@ -298,40 +378,70 @@ func testIterativeRandomStateSync(t *testing.T, count int) {
dstDb := rawdb.NewMemoryDatabase() dstDb := rawdb.NewMemoryDatabase()
sched := NewStateSync(srcRoot, dstDb, nil) sched := NewStateSync(srcRoot, dstDb, nil)
queue := make(map[common.Hash]struct{}) nodeQueue := make(map[string]stateElement)
nodes, _, codes := sched.Missing(count) codeQueue := make(map[common.Hash]struct{})
for _, hash := range append(nodes, codes...) { paths, nodes, codes := sched.Missing(count)
queue[hash] = struct{}{} for i, path := range paths {
nodeQueue[path] = stateElement{
path: path,
hash: nodes[i],
syncPath: trie.NewSyncPath([]byte(path)),
}
} }
for len(queue) > 0 { for _, hash := range codes {
codeQueue[hash] = struct{}{}
}
for len(nodeQueue)+len(codeQueue) > 0 {
// Fetch all the queued nodes in a random order // Fetch all the queued nodes in a random order
results := make([]trie.SyncResult, 0, len(queue)) if len(codeQueue) > 0 {
for hash := range queue { results := make([]trie.CodeSyncResult, 0, len(codeQueue))
data, err := srcDb.TrieDB().Node(hash) for hash := range codeQueue {
if err != nil { data, err := srcDb.ContractCode(common.Hash{}, hash)
data, err = srcDb.ContractCode(common.Hash{}, hash) if err != nil {
t.Fatalf("failed to retrieve node data for %x", hash)
}
results = append(results, trie.CodeSyncResult{Hash: hash, Data: data})
} }
if err != nil { for _, result := range results {
t.Fatalf("failed to retrieve node data for %x", hash) if err := sched.ProcessCode(result); err != nil {
t.Fatalf("failed to process result %v", err)
}
}
}
if len(nodeQueue) > 0 {
results := make([]trie.NodeSyncResult, 0, len(nodeQueue))
for path, element := range nodeQueue {
data, err := srcDb.TrieDB().Node(element.hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x %v %v", element.hash, []byte(element.path), element.path)
}
results = append(results, trie.NodeSyncResult{Path: path, Data: data})
}
for _, result := range results {
if err := sched.ProcessNode(result); err != nil {
t.Fatalf("failed to process result %v", err)
}
} }
results = append(results, trie.SyncResult{Hash: hash, Data: data})
} }
// Feed the retrieved results back and queue new tasks // Feed the retrieved results back and queue new tasks
for _, result := range results {
if err := sched.Process(result); err != nil {
t.Fatalf("failed to process result %v", err)
}
}
batch := dstDb.NewBatch() batch := dstDb.NewBatch()
if err := sched.Commit(batch); err != nil { if err := sched.Commit(batch); err != nil {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write() batch.Write()
queue = make(map[common.Hash]struct{}) nodeQueue = make(map[string]stateElement)
nodes, _, codes = sched.Missing(count) codeQueue = make(map[common.Hash]struct{})
for _, hash := range append(nodes, codes...) { paths, nodes, codes := sched.Missing(count)
queue[hash] = struct{}{} for i, path := range paths {
nodeQueue[path] = stateElement{
path: path,
hash: nodes[i],
syncPath: trie.NewSyncPath([]byte(path)),
}
}
for _, hash := range codes {
codeQueue[hash] = struct{}{}
} }
} }
// Cross check that the two states are in sync // Cross check that the two states are in sync
@ -348,34 +458,62 @@ func TestIterativeRandomDelayedStateSync(t *testing.T) {
dstDb := rawdb.NewMemoryDatabase() dstDb := rawdb.NewMemoryDatabase()
sched := NewStateSync(srcRoot, dstDb, nil) sched := NewStateSync(srcRoot, dstDb, nil)
queue := make(map[common.Hash]struct{}) nodeQueue := make(map[string]stateElement)
nodes, _, codes := sched.Missing(0) codeQueue := make(map[common.Hash]struct{})
for _, hash := range append(nodes, codes...) { paths, nodes, codes := sched.Missing(0)
queue[hash] = struct{}{} for i, path := range paths {
nodeQueue[path] = stateElement{
path: path,
hash: nodes[i],
syncPath: trie.NewSyncPath([]byte(path)),
}
} }
for len(queue) > 0 { for _, hash := range codes {
codeQueue[hash] = struct{}{}
}
for len(nodeQueue)+len(codeQueue) > 0 {
// Sync only half of the scheduled nodes, even those in random order // Sync only half of the scheduled nodes, even those in random order
results := make([]trie.SyncResult, 0, len(queue)/2+1) if len(codeQueue) > 0 {
for hash := range queue { results := make([]trie.CodeSyncResult, 0, len(codeQueue)/2+1)
delete(queue, hash) for hash := range codeQueue {
delete(codeQueue, hash)
data, err := srcDb.TrieDB().Node(hash) data, err := srcDb.ContractCode(common.Hash{}, hash)
if err != nil { if err != nil {
data, err = srcDb.ContractCode(common.Hash{}, hash) t.Fatalf("failed to retrieve node data for %x", hash)
} }
if err != nil { results = append(results, trie.CodeSyncResult{Hash: hash, Data: data})
t.Fatalf("failed to retrieve node data for %x", hash)
}
results = append(results, trie.SyncResult{Hash: hash, Data: data})
if len(results) >= cap(results) { if len(results) >= cap(results) {
break break
}
}
for _, result := range results {
if err := sched.ProcessCode(result); err != nil {
t.Fatalf("failed to process result %v", err)
}
} }
} }
// Feed the retrieved results back and queue new tasks if len(nodeQueue) > 0 {
for _, result := range results { results := make([]trie.NodeSyncResult, 0, len(nodeQueue)/2+1)
if err := sched.Process(result); err != nil { for path, element := range nodeQueue {
t.Fatalf("failed to process result %v", err) delete(nodeQueue, path)
data, err := srcDb.TrieDB().Node(element.hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x", element.hash)
}
results = append(results, trie.NodeSyncResult{Path: path, Data: data})
if len(results) >= cap(results) {
break
}
}
// Feed the retrieved results back and queue new tasks
for _, result := range results {
if err := sched.ProcessNode(result); err != nil {
t.Fatalf("failed to process result %v", err)
}
} }
} }
batch := dstDb.NewBatch() batch := dstDb.NewBatch()
@ -383,12 +521,17 @@ func TestIterativeRandomDelayedStateSync(t *testing.T) {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write() batch.Write()
for _, result := range results {
delete(queue, result.Hash) paths, nodes, codes := sched.Missing(0)
for i, path := range paths {
nodeQueue[path] = stateElement{
path: path,
hash: nodes[i],
syncPath: trie.NewSyncPath([]byte(path)),
}
} }
nodes, _, codes = sched.Missing(0) for _, hash := range codes {
for _, hash := range append(nodes, codes...) { codeQueue[hash] = struct{}{}
queue[hash] = struct{}{}
} }
} }
// Cross check that the two states are in sync // Cross check that the two states are in sync
@ -415,28 +558,62 @@ func TestIncompleteStateSync(t *testing.T) {
dstDb := rawdb.NewMemoryDatabase() dstDb := rawdb.NewMemoryDatabase()
sched := NewStateSync(srcRoot, dstDb, nil) sched := NewStateSync(srcRoot, dstDb, nil)
var added []common.Hash var (
addedCodes []common.Hash
nodes, _, codes := sched.Missing(1) addedNodes []common.Hash
queue := append(append([]common.Hash{}, nodes...), codes...) )
nodeQueue := make(map[string]stateElement)
for len(queue) > 0 { codeQueue := make(map[common.Hash]struct{})
// Fetch a batch of state nodes paths, nodes, codes := sched.Missing(1)
results := make([]trie.SyncResult, len(queue)) for i, path := range paths {
for i, hash := range queue { nodeQueue[path] = stateElement{
data, err := srcDb.TrieDB().Node(hash) path: path,
if err != nil { hash: nodes[i],
data, err = srcDb.ContractCode(common.Hash{}, hash) syncPath: trie.NewSyncPath([]byte(path)),
}
if err != nil {
t.Fatalf("failed to retrieve node data for %x", hash)
}
results[i] = trie.SyncResult{Hash: hash, Data: data}
} }
// Process each of the state nodes }
for _, result := range results { for _, hash := range codes {
if err := sched.Process(result); err != nil { codeQueue[hash] = struct{}{}
t.Fatalf("failed to process result %v", err) }
for len(nodeQueue)+len(codeQueue) > 0 {
// Fetch a batch of state nodes
if len(codeQueue) > 0 {
results := make([]trie.CodeSyncResult, 0, len(codeQueue))
for hash := range codeQueue {
data, err := srcDb.ContractCode(common.Hash{}, hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x", hash)
}
results = append(results, trie.CodeSyncResult{Hash: hash, Data: data})
addedCodes = append(addedCodes, hash)
}
// Process each of the state nodes
for _, result := range results {
if err := sched.ProcessCode(result); err != nil {
t.Fatalf("failed to process result %v", err)
}
}
}
var nodehashes []common.Hash
if len(nodeQueue) > 0 {
results := make([]trie.NodeSyncResult, 0, len(nodeQueue))
for key, element := range nodeQueue {
data, err := srcDb.TrieDB().Node(element.hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x", element.hash)
}
results = append(results, trie.NodeSyncResult{Path: key, Data: data})
if element.hash != srcRoot {
addedNodes = append(addedNodes, element.hash)
}
nodehashes = append(nodehashes, element.hash)
}
// Process each of the state nodes
for _, result := range results {
if err := sched.ProcessNode(result); err != nil {
t.Fatalf("failed to process result %v", err)
}
} }
} }
batch := dstDb.NewBatch() batch := dstDb.NewBatch()
@ -444,43 +621,44 @@ func TestIncompleteStateSync(t *testing.T) {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write() batch.Write()
for _, result := range results {
added = append(added, result.Hash) for _, root := range nodehashes {
// Check that all known sub-tries added so far are complete or missing entirely.
if _, ok := isCode[result.Hash]; ok {
continue
}
// Can't use checkStateConsistency here because subtrie keys may have odd // Can't use checkStateConsistency here because subtrie keys may have odd
// length and crash in LeafKey. // length and crash in LeafKey.
if err := checkTrieConsistency(dstDb, result.Hash); err != nil { if err := checkTrieConsistency(dstDb, root); err != nil {
t.Fatalf("state inconsistent: %v", err) t.Fatalf("state inconsistent: %v", err)
} }
} }
// Fetch the next batch to retrieve // Fetch the next batch to retrieve
nodes, _, codes = sched.Missing(1) nodeQueue = make(map[string]stateElement)
queue = append(append(queue[:0], nodes...), codes...) codeQueue = make(map[common.Hash]struct{})
paths, nodes, codes := sched.Missing(1)
for i, path := range paths {
nodeQueue[path] = stateElement{
path: path,
hash: nodes[i],
syncPath: trie.NewSyncPath([]byte(path)),
}
}
for _, hash := range codes {
codeQueue[hash] = struct{}{}
}
} }
// Sanity check that removing any node from the database is detected // Sanity check that removing any node from the database is detected
for _, node := range added[1:] { for _, node := range addedCodes {
var ( val := rawdb.ReadCode(dstDb, node)
key = node.Bytes() rawdb.DeleteCode(dstDb, node)
_, code = isCode[node] if err := checkStateConsistency(dstDb, srcRoot); err == nil {
val []byte t.Errorf("trie inconsistency not caught, missing: %x", node)
)
if code {
val = rawdb.ReadCode(dstDb, node)
rawdb.DeleteCode(dstDb, node)
} else {
val = rawdb.ReadTrieNode(dstDb, node)
rawdb.DeleteTrieNode(dstDb, node)
} }
if err := checkStateConsistency(dstDb, added[0]); err == nil { rawdb.WriteCode(dstDb, node, val)
t.Fatalf("trie inconsistency not caught, missing: %x", key) }
} for _, node := range addedNodes {
if code { val := rawdb.ReadTrieNode(dstDb, node)
rawdb.WriteCode(dstDb, node, val) rawdb.DeleteTrieNode(dstDb, node)
} else { if err := checkStateConsistency(dstDb, srcRoot); err == nil {
rawdb.WriteTrieNode(dstDb, node, val) t.Errorf("trie inconsistency not caught, missing: %v", node.Hex())
} }
rawdb.WriteTrieNode(dstDb, node, val)
} }
} }

70
eth/downloader/statesync.go
View file

@ -35,7 +35,7 @@ import (
// a single data retrieval network packet. // a single data retrieval network packet.
type stateReq struct { type stateReq struct {
nItems uint16 // Number of items requested for download (max is 384, so uint16 is sufficient) nItems uint16 // Number of items requested for download (max is 384, so uint16 is sufficient)
trieTasks map[common.Hash]*trieTask // Trie node download tasks to track previous attempts trieTasks map[string]*trieTask // Trie node download tasks to track previous attempts
codeTasks map[common.Hash]*codeTask // Byte code download tasks to track previous attempts codeTasks map[common.Hash]*codeTask // Byte code download tasks to track previous attempts
timeout time.Duration // Maximum round trip time for this to complete timeout time.Duration // Maximum round trip time for this to complete
timer *time.Timer // Timer to fire when the RTT timeout expires timer *time.Timer // Timer to fire when the RTT timeout expires
@ -257,8 +257,8 @@ type stateSync struct {
sched *trie.Sync // State trie sync scheduler defining the tasks sched *trie.Sync // State trie sync scheduler defining the tasks
keccak hash.Hash // Keccak256 hasher to verify deliveries with keccak hash.Hash // Keccak256 hasher to verify deliveries with
trieTasks map[common.Hash]*trieTask // Set of trie node tasks currently queued for retrieval trieTasks map[string]*trieTask // Set of trie node tasks currently queued for retrieval, indexed by path
codeTasks map[common.Hash]*codeTask // Set of byte code tasks currently queued for retrieval codeTasks map[common.Hash]*codeTask // Set of byte code tasks currently queued for retrieval, indexed by hash
numUncommitted int numUncommitted int
bytesUncommitted int bytesUncommitted int
@ -277,6 +277,7 @@ type stateSync struct {
// trieTask represents a single trie node download task, containing a set of // trieTask represents a single trie node download task, containing a set of
// peers already attempted retrieval from to detect stalled syncs and abort. // peers already attempted retrieval from to detect stalled syncs and abort.
type trieTask struct { type trieTask struct {
hash common.Hash
path [][]byte path [][]byte
attempts map[string]struct{} attempts map[string]struct{}
} }
@ -295,7 +296,7 @@ func newStateSync(d *Downloader, root common.Hash) *stateSync {
d: d, d: d,
sched: state.NewStateSync(root, d.stateDB, nil), sched: state.NewStateSync(root, d.stateDB, nil),
keccak: sha3.NewLegacyKeccak256(), keccak: sha3.NewLegacyKeccak256(),
trieTasks: make(map[common.Hash]*trieTask), trieTasks: make(map[string]*trieTask),
codeTasks: make(map[common.Hash]*codeTask), codeTasks: make(map[common.Hash]*codeTask),
deliver: make(chan *stateReq), deliver: make(chan *stateReq),
cancel: make(chan struct{}), cancel: make(chan struct{}),
@ -446,10 +447,11 @@ func (s *stateSync) assignTasks() {
func (s *stateSync) fillTasks(n int, req *stateReq) (nodes []common.Hash, paths []trie.SyncPath, codes []common.Hash) { func (s *stateSync) fillTasks(n int, req *stateReq) (nodes []common.Hash, paths []trie.SyncPath, codes []common.Hash) {
// Refill available tasks from the scheduler. // Refill available tasks from the scheduler.
if fill := n - (len(s.trieTasks) + len(s.codeTasks)); fill > 0 { if fill := n - (len(s.trieTasks) + len(s.codeTasks)); fill > 0 {
nodes, paths, codes := s.sched.Missing(fill) paths, hashes, codes := s.sched.Missing(fill)
for i, hash := range nodes { for i, path := range paths {
s.trieTasks[hash] = &trieTask{ s.trieTasks[path] = &trieTask{
path: paths[i], hash: hashes[i],
path: trie.NewSyncPath([]byte(path)),
attempts: make(map[string]struct{}), attempts: make(map[string]struct{}),
} }
} }
@ -465,7 +467,7 @@ func (s *stateSync) fillTasks(n int, req *stateReq) (nodes []common.Hash, paths
paths = make([]trie.SyncPath, 0, n) paths = make([]trie.SyncPath, 0, n)
codes = make([]common.Hash, 0, n) codes = make([]common.Hash, 0, n)
req.trieTasks = make(map[common.Hash]*trieTask, n) req.trieTasks = make(map[string]*trieTask, n)
req.codeTasks = make(map[common.Hash]*codeTask, n) req.codeTasks = make(map[common.Hash]*codeTask, n)
for hash, t := range s.codeTasks { for hash, t := range s.codeTasks {
@ -483,7 +485,7 @@ func (s *stateSync) fillTasks(n int, req *stateReq) (nodes []common.Hash, paths
req.codeTasks[hash] = t req.codeTasks[hash] = t
delete(s.codeTasks, hash) delete(s.codeTasks, hash)
} }
for hash, t := range s.trieTasks { for path, t := range s.trieTasks {
// Stop when we've gathered enough requests // Stop when we've gathered enough requests
if len(nodes)+len(codes) == n { if len(nodes)+len(codes) == n {
break break
@ -495,11 +497,11 @@ func (s *stateSync) fillTasks(n int, req *stateReq) (nodes []common.Hash, paths
// Assign the request to this peer // Assign the request to this peer
t.attempts[req.peer.id] = struct{}{} t.attempts[req.peer.id] = struct{}{}
nodes = append(nodes, hash) nodes = append(nodes, t.hash)
paths = append(paths, t.path) paths = append(paths, t.path)
req.trieTasks[hash] = t req.trieTasks[path] = t
delete(s.trieTasks, hash) delete(s.trieTasks, path)
} }
req.nItems = uint16(len(nodes) + len(codes)) req.nItems = uint16(len(nodes) + len(codes))
return nodes, paths, codes return nodes, paths, codes
@ -521,7 +523,7 @@ func (s *stateSync) process(req *stateReq) (int, error) {
// Iterate over all the delivered data and inject one-by-one into the trie // Iterate over all the delivered data and inject one-by-one into the trie
for _, blob := range req.response { for _, blob := range req.response {
hash, err := s.processNodeData(blob) hash, err := s.processNodeData(req.trieTasks, req.codeTasks, blob)
switch err { switch err {
case nil: case nil:
s.numUncommitted++ s.numUncommitted++
@ -534,13 +536,10 @@ func (s *stateSync) process(req *stateReq) (int, error) {
default: default:
return successful, fmt.Errorf("invalid state node %s: %v", hash.TerminalString(), err) return successful, fmt.Errorf("invalid state node %s: %v", hash.TerminalString(), err)
} }
// Delete from both queues (one delivery is enough for the syncer)
delete(req.trieTasks, hash)
delete(req.codeTasks, hash)
} }
// Put unfulfilled tasks back into the retry queue // Put unfulfilled tasks back into the retry queue
npeers := s.d.peers.Len() npeers := s.d.peers.Len()
for hash, task := range req.trieTasks { for path, task := range req.trieTasks {
// If the node did deliver something, missing items may be due to a protocol // If the node did deliver something, missing items may be due to a protocol
// limit or a previous timeout + delayed delivery. Both cases should permit // limit or a previous timeout + delayed delivery. Both cases should permit
// the node to retry the missing items (to avoid single-peer stalls). // the node to retry the missing items (to avoid single-peer stalls).
@ -550,10 +549,10 @@ func (s *stateSync) process(req *stateReq) (int, error) {
// If we've requested the node too many times already, it may be a malicious // If we've requested the node too many times already, it may be a malicious
// sync where nobody has the right data. Abort. // sync where nobody has the right data. Abort.
if len(task.attempts) >= npeers { if len(task.attempts) >= npeers {
return successful, fmt.Errorf("trie node %s failed with all peers (%d tries, %d peers)", hash.TerminalString(), len(task.attempts), npeers) return successful, fmt.Errorf("trie node %s failed with all peers (%d tries, %d peers)", task.hash.TerminalString(), len(task.attempts), npeers)
} }
// Missing item, place into the retry queue. // Missing item, place into the retry queue.
s.trieTasks[hash] = task s.trieTasks[path] = task
} }
for hash, task := range req.codeTasks { for hash, task := range req.codeTasks {
// If the node did deliver something, missing items may be due to a protocol // If the node did deliver something, missing items may be due to a protocol
@ -576,13 +575,34 @@ func (s *stateSync) process(req *stateReq) (int, error) {
// processNodeData tries to inject a trie node data blob delivered from a remote // processNodeData tries to inject a trie node data blob delivered from a remote
// peer into the state trie, returning whether anything useful was written or any // peer into the state trie, returning whether anything useful was written or any
// error occurred. // error occurred.
func (s *stateSync) processNodeData(blob []byte) (common.Hash, error) { //
res := trie.SyncResult{Data: blob} // If multiple requests correspond to the same hash, this method will inject the
// blob as a result for the first one only, leaving the remaining duplicates to
// be fetched again.
func (s *stateSync) processNodeData(nodeTasks map[string]*trieTask, codeTasks map[common.Hash]*codeTask, blob []byte) (common.Hash, error) {
s.keccak.Reset() s.keccak.Reset()
s.keccak.Write(blob) s.keccak.Write(blob)
s.keccak.Sum(res.Hash[:0]) hash := common.BytesToHash(s.keccak.Sum(nil))
err := s.sched.Process(res)
return res.Hash, err if _, present := codeTasks[hash]; present {
err := s.sched.ProcessCode(trie.CodeSyncResult{
Hash: hash,
Data: blob,
})
delete(codeTasks, hash)
return hash, err
}
for path, task := range nodeTasks {
if task.hash == hash {
err := s.sched.ProcessNode(trie.NodeSyncResult{
Path: path,
Data: blob,
})
delete(nodeTasks, path)
return hash, err
}
}
return common.Hash{}, trie.ErrNotRequested
} }
// updateStats bumps the various state sync progress counters and displays a log // updateStats bumps the various state sync progress counters and displays a log

34
trie/committer.go
View file

@ -32,7 +32,8 @@ const leafChanSize = 200
type leaf struct { type leaf struct {
size int // size of the rlp data (estimate) size int // size of the rlp data (estimate)
hash common.Hash // hash of rlp data hash common.Hash // hash of rlp data
node node // the Node to commit node node // the node to commit
path []byte // the path from the root node
} }
// committer is a type used for the trie Commit operation. A committer has some // committer is a type used for the trie Commit operation. A committer has some
@ -69,15 +70,15 @@ func (c *committer) Commit(n node, db *Database) (hashNode, int, error) {
if db == nil { if db == nil {
return nil, 0, errors.New("no Db provided") return nil, 0, errors.New("no Db provided")
} }
h, committed, err := c.commit(n, db) h, committed, err := c.commit(nil, n, db)
if err != nil { if err != nil {
return nil, 0, err return nil, 0, err
} }
return h.(hashNode), committed, nil return h.(hashNode), committed, nil
} }
// commit collapses a Node down into a hash Node and inserts it into the database // commit collapses a node down into a hash node and inserts it into the database
func (c *committer) commit(n node, db *Database) (node, int, error) { func (c *committer) commit(path []byte, n node, db *Database) (node, int, error) {
// if this path is clean, use available cached data // if this path is clean, use available cached data
hash, dirty := n.cache() hash, dirty := n.cache()
if hash != nil && !dirty { if hash != nil && !dirty {
@ -93,7 +94,7 @@ func (c *committer) commit(n node, db *Database) (node, int, error) {
// otherwise it can only be hashNode or valueNode. // otherwise it can only be hashNode or valueNode.
var childCommitted int var childCommitted int
if _, ok := cn.Val.(*fullNode); ok { if _, ok := cn.Val.(*fullNode); ok {
childV, committed, err := c.commit(cn.Val, db) childV, committed, err := c.commit(append(path, cn.Key...), cn.Val, db)
if err != nil { if err != nil {
return nil, 0, err return nil, 0, err
} }
@ -101,20 +102,20 @@ func (c *committer) commit(n node, db *Database) (node, int, error) {
} }
// The key needs to be copied, since we're delivering it to database // The key needs to be copied, since we're delivering it to database
collapsed.Key = hexToCompact(cn.Key) collapsed.Key = hexToCompact(cn.Key)
hashedNode := c.store(collapsed, db) hashedNode := c.store(path, collapsed, db)
if hn, ok := hashedNode.(hashNode); ok { if hn, ok := hashedNode.(hashNode); ok {
return hn, childCommitted + 1, nil return hn, childCommitted + 1, nil
} }
return collapsed, childCommitted, nil return collapsed, childCommitted, nil
case *fullNode: case *fullNode:
hashedKids, childCommitted, err := c.commitChildren(cn, db) hashedKids, childCommitted, err := c.commitChildren(path, cn, db)
if err != nil { if err != nil {
return nil, 0, err return nil, 0, err
} }
collapsed := cn.copy() collapsed := cn.copy()
collapsed.Children = hashedKids collapsed.Children = hashedKids
hashedNode := c.store(collapsed, db) hashedNode := c.store(path, collapsed, db)
if hn, ok := hashedNode.(hashNode); ok { if hn, ok := hashedNode.(hashNode); ok {
return hn, childCommitted + 1, nil return hn, childCommitted + 1, nil
} }
@ -128,7 +129,7 @@ func (c *committer) commit(n node, db *Database) (node, int, error) {
} }
// commitChildren commits the children of the given fullnode // commitChildren commits the children of the given fullnode
func (c *committer) commitChildren(n *fullNode, db *Database) ([17]node, int, error) { func (c *committer) commitChildren(path []byte, n *fullNode, db *Database) ([17]node, int, error) {
var ( var (
committed int committed int
children [17]node children [17]node
@ -148,7 +149,7 @@ func (c *committer) commitChildren(n *fullNode, db *Database) ([17]node, int, er
// Commit the child recursively and store the "hashed" value. // Commit the child recursively and store the "hashed" value.
// Note the returned node can be some embedded nodes, so it's // Note the returned node can be some embedded nodes, so it's
// possible the type is not hashNode. // possible the type is not hashNode.
hashed, childCommitted, err := c.commit(child, db) hashed, childCommitted, err := c.commit(append(path, byte(i)), child, db)
if err != nil { if err != nil {
return children, 0, err return children, 0, err
} }
@ -162,10 +163,10 @@ func (c *committer) commitChildren(n *fullNode, db *Database) ([17]node, int, er
return children, committed, nil return children, committed, nil
} }
// store hashes the Node n and if we have a storage layer specified, it writes // store hashes the node n and if we have a storage layer specified, it writes
// the key/value pair to it and tracks any Node->child references as well as any // the key/value pair to it and tracks any node->child references as well as any
// Node->external trie references. // node->external trie references.
func (c *committer) store(n node, db *Database) node { func (c *committer) store(path []byte, n node, db *Database) node {
// Larger nodes are replaced by their hash and stored in the database. // Larger nodes are replaced by their hash and stored in the database.
var ( var (
hash, _ = n.cache() hash, _ = n.cache()
@ -189,6 +190,7 @@ func (c *committer) store(n node, db *Database) node {
size: size, size: size,
hash: common.BytesToHash(hash), hash: common.BytesToHash(hash),
node: n, node: n,
path: path,
} }
} else if db != nil { } else if db != nil {
// No leaf-callback used, but there's still a database. Do serial // No leaf-callback used, but there's still a database. Do serial
@ -213,13 +215,13 @@ func (c *committer) commitLoop(db *Database) {
switch n := n.(type) { switch n := n.(type) {
case *shortNode: case *shortNode:
if child, ok := n.Val.(valueNode); ok { if child, ok := n.Val.(valueNode); ok {
c.onleaf(nil, nil, child, hash) c.onleaf(nil, nil, child, hash, nil)
} }
case *fullNode: case *fullNode:
// For children in range [0, 15], it's impossible // For children in range [0, 15], it's impossible
// to contain valueNode. Only check the 17th child. // to contain valueNode. Only check the 17th child.
if n.Children[16] != nil { if n.Children[16] != nil {
c.onleaf(nil, nil, n.Children[16].(valueNode), hash) c.onleaf(nil, nil, n.Children[16].(valueNode), hash, nil)
} }
} }
} }

331
trie/sync.go
View file

@ -25,6 +25,7 @@ import (
"github.com/XinFinOrg/XDPoSChain/core/rawdb" "github.com/XinFinOrg/XDPoSChain/core/rawdb"
"github.com/XinFinOrg/XDPoSChain/core/types" "github.com/XinFinOrg/XDPoSChain/core/types"
"github.com/XinFinOrg/XDPoSChain/ethdb" "github.com/XinFinOrg/XDPoSChain/ethdb"
"github.com/XinFinOrg/XDPoSChain/log"
) )
// ErrNotRequested is returned by the trie sync when it's requested to process a // ErrNotRequested is returned by the trie sync when it's requested to process a
@ -40,19 +41,6 @@ var ErrAlreadyProcessed = errors.New("already processed")
// memory if the node was configured with a significant number of peers. // memory if the node was configured with a significant number of peers.
const maxFetchesPerDepth = 16384 const maxFetchesPerDepth = 16384
// request represents a scheduled or already in-flight state retrieval request.
type request struct {
path []byte // Merkle path leading to this node for prioritization
hash common.Hash // Hash of the Node data content to retrieve
data []byte // Data content of the Node, cached until all subtrees complete
code bool // Whether this is a code entry
parents []*request // Parent state nodes referencing this entry (notify all upon completion)
deps int // Number of dependencies before allowed to commit this Node
callback LeafCallback // Callback to invoke if a leaf Node it reached on this branch
}
// SyncPath is a path tuple identifying a particular trie node either in a single // SyncPath is a path tuple identifying a particular trie node either in a single
// trie (account) or a layered trie (account -> storage). // trie (account) or a layered trie (account -> storage).
// //
@ -72,9 +60,9 @@ type request struct {
// - Path 0x012345678901234567890123456789010123456789012345678901234567890199 -> {0x0123456789012345678901234567890101234567890123456789012345678901, 0x0099} // - Path 0x012345678901234567890123456789010123456789012345678901234567890199 -> {0x0123456789012345678901234567890101234567890123456789012345678901, 0x0099}
type SyncPath [][]byte type SyncPath [][]byte
// newSyncPath converts an expanded trie path from nibble form into a compact // NewSyncPath converts an expanded trie path from nibble form into a compact
// version that can be sent over the network. // version that can be sent over the network.
func newSyncPath(path []byte) SyncPath { func NewSyncPath(path []byte) SyncPath {
// If the hash is from the account trie, append a single item, if it // If the hash is from the account trie, append a single item, if it
// is from the a storage trie, append a tuple. Note, the length 64 is // is from the a storage trie, append a tuple. Note, the length 64 is
// clashing between account leaf and storage root. It's fine though // clashing between account leaf and storage root. It's fine though
@ -86,30 +74,57 @@ func newSyncPath(path []byte) SyncPath {
return SyncPath{hexToKeybytes(path[:64]), hexToCompact(path[64:])} return SyncPath{hexToKeybytes(path[:64]), hexToCompact(path[64:])}
} }
// SyncResult is a response with requested data along with it's hash. // nodeRequest represents a scheduled or already in-flight trie node retrieval request.
type SyncResult struct { type nodeRequest struct {
Hash common.Hash // Hash of the originally unknown trie Node hash common.Hash // Hash of the trie node to retrieve
Data []byte // Data content of the retrieved Node path []byte // Merkle path leading to this node for prioritization
data []byte // Data content of the node, cached until all subtrees complete
parent *nodeRequest // Parent state node referencing this entry
deps int // Number of dependencies before allowed to commit this node
callback LeafCallback // Callback to invoke if a leaf node it reached on this branch
}
// codeRequest represents a scheduled or already in-flight bytecode retrieval request.
type codeRequest struct {
hash common.Hash // Hash of the contract bytecode to retrieve
path []byte // Merkle path leading to this node for prioritization
data []byte // Data content of the node, cached until all subtrees complete
parents []*nodeRequest // Parent state nodes referencing this entry (notify all upon completion)
}
// NodeSyncResult is a response with requested trie node along with its node path.
type NodeSyncResult struct {
Path string // Path of the originally unknown trie node
Data []byte // Data content of the retrieved trie node
}
// CodeSyncResult is a response with requested bytecode along with its hash.
type CodeSyncResult struct {
Hash common.Hash // Hash the originally unknown bytecode
Data []byte // Data content of the retrieved bytecode
} }
// syncMemBatch is an in-memory buffer of successfully downloaded but not yet // syncMemBatch is an in-memory buffer of successfully downloaded but not yet
// persisted data items. // persisted data items.
type syncMemBatch struct { type syncMemBatch struct {
nodes map[common.Hash][]byte // In-memory membatch of recently completed nodes nodes map[string][]byte // In-memory membatch of recently completed nodes
codes map[common.Hash][]byte // In-memory membatch of recently completed codes hashes map[string]common.Hash // Hashes of recently completed nodes
codes map[common.Hash][]byte // In-memory membatch of recently completed codes
} }
// newSyncMemBatch allocates a new memory-buffer for not-yet persisted trie nodes. // newSyncMemBatch allocates a new memory-buffer for not-yet persisted trie nodes.
func newSyncMemBatch() *syncMemBatch { func newSyncMemBatch() *syncMemBatch {
return &syncMemBatch{ return &syncMemBatch{
nodes: make(map[common.Hash][]byte), nodes: make(map[string][]byte),
codes: make(map[common.Hash][]byte), hashes: make(map[string]common.Hash),
codes: make(map[common.Hash][]byte),
} }
} }
// hasNode reports the trie node with specific hash is already cached. // hasNode reports the trie node with specific path is already cached.
func (batch *syncMemBatch) hasNode(hash common.Hash) bool { func (batch *syncMemBatch) hasNode(path []byte) bool {
_, ok := batch.nodes[hash] _, ok := batch.nodes[string(path)]
return ok return ok
} }
@ -123,12 +138,12 @@ func (batch *syncMemBatch) hasCode(hash common.Hash) bool {
// unknown trie hashes to retrieve, accepts Node data associated with said hashes // unknown trie hashes to retrieve, accepts Node data associated with said hashes
// and reconstructs the trie step by step until all is done. // and reconstructs the trie step by step until all is done.
type Sync struct { type Sync struct {
database ethdb.KeyValueReader // Persistent database to check for existing entries database ethdb.KeyValueReader // Persistent database to check for existing entries
membatch *syncMemBatch // Memory buffer to avoid frequent database writes membatch *syncMemBatch // Memory buffer to avoid frequent database writes
nodeReqs map[common.Hash]*request // Pending requests pertaining to a trie node hash nodeReqs map[string]*nodeRequest // Pending requests pertaining to a trie node path
codeReqs map[common.Hash]*request // Pending requests pertaining to a code hash codeReqs map[common.Hash]*codeRequest // Pending requests pertaining to a code hash
queue *prque.Prque[int64, any] // Priority queue with the pending requests queue *prque.Prque[int64, any] // Priority queue with the pending requests
fetches map[int]int // Number of active fetches per trie node depth fetches map[int]int // Number of active fetches per trie node depth
} }
// NewSync creates a new trie data download scheduler. // NewSync creates a new trie data download scheduler.
@ -136,51 +151,51 @@ func NewSync(root common.Hash, database ethdb.KeyValueReader, callback LeafCallb
ts := &Sync{ ts := &Sync{
database: database, database: database,
membatch: newSyncMemBatch(), membatch: newSyncMemBatch(),
nodeReqs: make(map[common.Hash]*request), nodeReqs: make(map[string]*nodeRequest),
codeReqs: make(map[common.Hash]*request), codeReqs: make(map[common.Hash]*codeRequest),
queue: prque.New[int64, any](nil), // Ugh, can contain both string and hash, whyyy queue: prque.New[int64, any](nil),
fetches: make(map[int]int), fetches: make(map[int]int),
} }
ts.AddSubTrie(root, nil, common.Hash{}, callback) ts.AddSubTrie(root, nil, common.Hash{}, nil, callback)
return ts return ts
} }
// AddSubTrie registers a new trie to the sync code, rooted at the designated parent. // AddSubTrie registers a new trie to the sync code, rooted at the designated
func (s *Sync) AddSubTrie(root common.Hash, path []byte, parent common.Hash, callback LeafCallback) { // parent for completion tracking. The given path is a unique node path in
// hex format and contain all the parent path if it's layered trie node.
func (s *Sync) AddSubTrie(root common.Hash, path []byte, parent common.Hash, parentPath []byte, callback LeafCallback) {
// Short circuit if the trie is empty or already known // Short circuit if the trie is empty or already known
if root == types.EmptyRootHash { if root == types.EmptyRootHash {
return return
} }
if s.membatch.hasNode(root) { if s.membatch.hasNode(path) {
return return
} }
// If database says this is a duplicate, then at least the trie node is
// present, and we hold the assumption that it's NOT legacy contract code.
if rawdb.HasTrieNode(s.database, root) { if rawdb.HasTrieNode(s.database, root) {
return return
} }
// Assemble the new sub-trie sync request // Assemble the new sub-trie sync request
req := &request{ req := &nodeRequest{
path: path,
hash: root, hash: root,
path: path,
callback: callback, callback: callback,
} }
// If this sub-trie has a designated parent, link them together // If this sub-trie has a designated parent, link them together
if parent != (common.Hash{}) { if parent != (common.Hash{}) {
ancestor := s.nodeReqs[parent] ancestor := s.nodeReqs[string(parentPath)]
if ancestor == nil { if ancestor == nil {
panic(fmt.Sprintf("sub-trie ancestor not found: %x", parent)) panic(fmt.Sprintf("sub-trie ancestor not found: %x", parent))
} }
ancestor.deps++ ancestor.deps++
req.parents = append(req.parents, ancestor) req.parent = ancestor
} }
s.schedule(req) s.scheduleNodeRequest(req)
} }
// AddCodeEntry schedules the direct retrieval of a contract code that should not // AddCodeEntry schedules the direct retrieval of a contract code that should not
// be interpreted as a trie node, but rather accepted and stored into the database // be interpreted as a trie node, but rather accepted and stored into the database
// as is. // as is.
func (s *Sync) AddCodeEntry(hash common.Hash, path []byte, parent common.Hash) { func (s *Sync) AddCodeEntry(hash common.Hash, path []byte, parent common.Hash, parentPath []byte) {
// Short circuit if the entry is empty or already known // Short circuit if the entry is empty or already known
if hash == types.EmptyCodeHash { if hash == types.EmptyCodeHash {
return return
@ -197,30 +212,29 @@ func (s *Sync) AddCodeEntry(hash common.Hash, path []byte, parent common.Hash) {
return return
} }
// Assemble the new sub-trie sync request // Assemble the new sub-trie sync request
req := &request{ req := &codeRequest{
path: path, path: path,
hash: hash, hash: hash,
code: true,
} }
// If this sub-trie has a designated parent, link them together // If this sub-trie has a designated parent, link them together
if parent != (common.Hash{}) { if parent != (common.Hash{}) {
ancestor := s.nodeReqs[parent] // the parent of codereq can ONLY be nodereq ancestor := s.nodeReqs[string(parentPath)] // the parent of codereq can ONLY be nodereq
if ancestor == nil { if ancestor == nil {
panic(fmt.Sprintf("raw-entry ancestor not found: %x", parent)) panic(fmt.Sprintf("raw-entry ancestor not found: %x", parent))
} }
ancestor.deps++ ancestor.deps++
req.parents = append(req.parents, ancestor) req.parents = append(req.parents, ancestor)
} }
s.schedule(req) s.scheduleCodeRequest(req)
} }
// Missing retrieves the known missing nodes from the trie for retrieval. To aid // Missing retrieves the known missing nodes from the trie for retrieval. To aid
// both eth/6x style fast sync and snap/1x style state sync, the paths of trie // both eth/6x style fast sync and snap/1x style state sync, the paths of trie
// nodes are returned too, as well as separate hash list for codes. // nodes are returned too, as well as separate hash list for codes.
func (s *Sync) Missing(max int) (nodes []common.Hash, paths []SyncPath, codes []common.Hash) { func (s *Sync) Missing(max int) ([]string, []common.Hash, []common.Hash) {
var ( var (
nodePaths []string
nodeHashes []common.Hash nodeHashes []common.Hash
nodePaths []SyncPath
codeHashes []common.Hash codeHashes []common.Hash
) )
for !s.queue.Empty() && (max == 0 || len(nodeHashes)+len(codeHashes) < max) { for !s.queue.Empty() && (max == 0 || len(nodeHashes)+len(codeHashes) < max) {
@ -236,62 +250,77 @@ func (s *Sync) Missing(max int) (nodes []common.Hash, paths []SyncPath, codes []
s.queue.Pop() s.queue.Pop()
s.fetches[depth]++ s.fetches[depth]++
hash := item.(common.Hash) switch item.(type) {
if req, ok := s.nodeReqs[hash]; ok { case common.Hash:
nodeHashes = append(nodeHashes, hash) codeHashes = append(codeHashes, item.(common.Hash))
nodePaths = append(nodePaths, newSyncPath(req.path)) case string:
} else { path := item.(string)
codeHashes = append(codeHashes, hash) req, ok := s.nodeReqs[path]
if !ok {
log.Error("Missing node request", "path", path)
continue // System very wrong, shouldn't happen
}
nodePaths = append(nodePaths, path)
nodeHashes = append(nodeHashes, req.hash)
} }
} }
return nodeHashes, nodePaths, codeHashes return nodePaths, nodeHashes, codeHashes
} }
// Process injects the received data for requested item. Note it can // ProcessCode injects the received data for requested item. Note it can
// happpen that the single response commits two pending requests(e.g. // happpen that the single response commits two pending requests(e.g.
// there are two requests one for code and one for node but the hash // there are two requests one for code and one for node but the hash
// is same). In this case the second response for the same hash will // is same). In this case the second response for the same hash will
// be treated as "non-requested" item or "already-processed" item but // be treated as "non-requested" item or "already-processed" item but
// there is no downside. // there is no downside.
func (s *Sync) Process(result SyncResult) error { func (s *Sync) ProcessCode(result CodeSyncResult) error {
// If the item was not requested either for code or node, bail out // If the code was not requested or it's already processed, bail out
if s.nodeReqs[result.Hash] == nil && s.codeReqs[result.Hash] == nil { req := s.codeReqs[result.Hash]
if req == nil {
return ErrNotRequested return ErrNotRequested
} }
// There is an pending code request for this data, commit directly if req.data != nil {
var filled bool
if req := s.codeReqs[result.Hash]; req != nil && req.data == nil {
filled = true
req.data = result.Data
s.commit(req)
}
// There is an pending node request for this data, fill it.
if req := s.nodeReqs[result.Hash]; req != nil && req.data == nil {
filled = true
// Decode the Node data content and update the request
node, err := decodeNode(result.Hash[:], result.Data)
if err != nil {
return err
}
req.data = result.Data
// Create and schedule a request for all the children nodes
requests, err := s.children(req, node)
if err != nil {
return err
}
if len(requests) == 0 && req.deps == 0 {
s.commit(req)
} else {
req.deps += len(requests)
for _, child := range requests {
s.schedule(child)
}
}
}
if !filled {
return ErrAlreadyProcessed return ErrAlreadyProcessed
} }
req.data = result.Data
return s.commitCodeRequest(req)
}
// ProcessNode injects the received data for requested item. Note it can
// happen that the single response commits two pending requests(e.g.
// there are two requests one for code and one for node but the hash
// is same). In this case the second response for the same hash will
// be treated as "non-requested" item or "already-processed" item but
// there is no downside.
func (s *Sync) ProcessNode(result NodeSyncResult) error {
// If the trie node was not requested or it's already processed, bail out
req := s.nodeReqs[result.Path]
if req == nil {
return ErrNotRequested
}
if req.data != nil {
return ErrAlreadyProcessed
}
// Decode the node data content and update the request
node, err := decodeNode(req.hash.Bytes(), result.Data)
if err != nil {
return err
}
req.data = result.Data
// Create and schedule a request for all the children nodes
requests, err := s.children(req, node)
if err != nil {
return err
}
if len(requests) == 0 && req.deps == 0 {
s.commitNodeRequest(req)
} else {
req.deps += len(requests)
for _, child := range requests {
s.scheduleNodeRequest(child)
}
}
return nil return nil
} }
@ -299,11 +328,11 @@ func (s *Sync) Process(result SyncResult) error {
// storage, returning any occurred error. // storage, returning any occurred error.
func (s *Sync) Commit(dbw ethdb.Batch) error { func (s *Sync) Commit(dbw ethdb.Batch) error {
// Dump the membatch into a database dbw // Dump the membatch into a database dbw
for key, value := range s.membatch.nodes { for path, value := range s.membatch.nodes {
rawdb.WriteTrieNode(dbw, key, value) rawdb.WriteTrieNode(dbw, s.membatch.hashes[path], value)
} }
for key, value := range s.membatch.codes { for hash, value := range s.membatch.codes {
rawdb.WriteCode(dbw, key, value) rawdb.WriteCode(dbw, hash, value)
} }
// Drop the membatch data and return // Drop the membatch data and return
s.membatch = newSyncMemBatch() s.membatch = newSyncMemBatch()
@ -318,23 +347,31 @@ func (s *Sync) Pending() int {
// schedule inserts a new state retrieval request into the fetch queue. If there // schedule inserts a new state retrieval request into the fetch queue. If there
// is already a pending request for this Node, the new request will be discarded // is already a pending request for this Node, the new request will be discarded
// and only a parent reference added to the old one. // and only a parent reference added to the old one.
func (s *Sync) schedule(req *request) { func (s *Sync) scheduleNodeRequest(req *nodeRequest) {
var reqset = s.nodeReqs s.nodeReqs[string(req.path)] = req
if req.code {
reqset = s.codeReqs // Schedule the request for future retrieval. This queue is shared
// by both node requests and code requests.
prio := int64(len(req.path)) << 56 // depth >= 128 will never happen, storage leaves will be included in their parents
for i := 0; i < 14 && i < len(req.path); i++ {
prio |= int64(15-req.path[i]) << (52 - i*4) // 15-nibble => lexicographic order
} }
s.queue.Push(string(req.path), prio)
}
// schedule inserts a new state retrieval request into the fetch queue. If there
// is already a pending request for this node, the new request will be discarded
// and only a parent reference added to the old one.
func (s *Sync) scheduleCodeRequest(req *codeRequest) {
// If we're already requesting this node, add a new reference and stop // If we're already requesting this node, add a new reference and stop
if old, ok := reqset[req.hash]; ok { if old, ok := s.codeReqs[req.hash]; ok {
old.parents = append(old.parents, req.parents...) old.parents = append(old.parents, req.parents...)
return return
} }
reqset[req.hash] = req s.codeReqs[req.hash] = req
// Schedule the request for future retrieval. This queue is shared // Schedule the request for future retrieval. This queue is shared
// by both node requests and code requests. It can happen that there // by both node requests and code requests.
// is a trie node and code has same hash. In this case two elements
// with same hash and same or different depth will be pushed. But it's
// ok the worst case is the second response will be treated as duplicated.
prio := int64(len(req.path)) << 56 // depth >= 128 will never happen, storage leaves will be included in their parents prio := int64(len(req.path)) << 56 // depth >= 128 will never happen, storage leaves will be included in their parents
for i := 0; i < 14 && i < len(req.path); i++ { for i := 0; i < 14 && i < len(req.path); i++ {
prio |= int64(15-req.path[i]) << (52 - i*4) // 15-nibble => lexicographic order prio |= int64(15-req.path[i]) << (52 - i*4) // 15-nibble => lexicographic order
@ -344,8 +381,8 @@ func (s *Sync) schedule(req *request) {
// children retrieves all the missing children of a state trie entry for future // children retrieves all the missing children of a state trie entry for future
// retrieval scheduling. // retrieval scheduling.
func (s *Sync) children(req *request, object node) ([]*request, error) { func (s *Sync) children(req *nodeRequest, object node) ([]*nodeRequest, error) {
// Gather all the children of the Node, irrelevant whether known or not // Gather all the children of the node, irrelevant whether known or not
type child struct { type child struct {
path []byte path []byte
node node node node
@ -375,7 +412,7 @@ func (s *Sync) children(req *request, object node) ([]*request, error) {
panic(fmt.Sprintf("unknown Node: %+v", node)) panic(fmt.Sprintf("unknown Node: %+v", node))
} }
// Iterate over the children, and request all unknown ones // Iterate over the children, and request all unknown ones
requests := make([]*request, 0, len(children)) requests := make([]*nodeRequest, 0, len(children))
for _, child := range children { for _, child := range children {
// Notify any external watcher of a new key/value Node // Notify any external watcher of a new key/value Node
if req.callback != nil { if req.callback != nil {
@ -387,28 +424,28 @@ func (s *Sync) children(req *request, object node) ([]*request, error) {
paths = append(paths, hexToKeybytes(child.path[:2*common.HashLength])) paths = append(paths, hexToKeybytes(child.path[:2*common.HashLength]))
paths = append(paths, hexToKeybytes(child.path[2*common.HashLength:])) paths = append(paths, hexToKeybytes(child.path[2*common.HashLength:]))
} }
if err := req.callback(paths, child.path, node, req.hash); err != nil { if err := req.callback(paths, child.path, node, req.hash, req.path); err != nil {
return nil, err return nil, err
} }
} }
} }
// If the child references another Node, resolve or schedule // If the child references another Node, resolve or schedule
if node, ok := (child.node).(hashNode); ok { if node, ok := (child.node).(hashNode); ok {
// Try to resolve the Node from the local database // Try to resolve the node from the local database
hash := common.BytesToHash(node) if s.membatch.hasNode(child.path) {
if s.membatch.hasNode(hash) {
continue continue
} }
// If database says duplicate, then at least the trie node is present // If database says duplicate, then at least the trie node is present
// and we hold the assumption that it's NOT legacy contract code. // and we hold the assumption that it's NOT legacy contract code.
if rawdb.HasTrieNode(s.database, hash) { chash := common.BytesToHash(node)
if rawdb.HasTrieNode(s.database, chash) {
continue continue
} }
// Locally unknown Node, schedule for retrieval // Locally unknown node, schedule for retrieval
requests = append(requests, &request{ requests = append(requests, &nodeRequest{
path: child.path, path: child.path,
hash: hash, hash: chash,
parents: []*request{req}, parent: req,
callback: req.callback, callback: req.callback,
}) })
} }
@ -419,22 +456,40 @@ func (s *Sync) children(req *request, object node) ([]*request, error) {
// commit finalizes a retrieval request and stores it into the membatch. If any // commit finalizes a retrieval request and stores it into the membatch. If any
// of the referencing parent requests complete due to this commit, they are also // of the referencing parent requests complete due to this commit, they are also
// committed themselves. // committed themselves.
func (s *Sync) commit(req *request) (err error) { func (s *Sync) commitNodeRequest(req *nodeRequest) error {
// Write the Node content to the membatch // Write the node content to the membatch
if req.code { s.membatch.nodes[string(req.path)] = req.data
s.membatch.codes[req.hash] = req.data s.membatch.hashes[string(req.path)] = req.hash
delete(s.codeReqs, req.hash)
s.fetches[len(req.path)]-- delete(s.nodeReqs, string(req.path))
} else { s.fetches[len(req.path)]--
s.membatch.nodes[req.hash] = req.data
delete(s.nodeReqs, req.hash) // Check parent for completion
s.fetches[len(req.path)]-- if req.parent != nil {
} req.parent.deps--
// Check all parents for completion if req.parent.deps == 0 {
for _, parent := range req.parents { if err := s.commitNodeRequest(req.parent); err != nil {
parent.deps-- return err
if parent.deps == 0 { }
if err := s.commit(parent); err != nil { }
}
return nil
}
// commit finalizes a retrieval request and stores it into the membatch. If any
// of the referencing parent requests complete due to this commit, they are also
// committed themselves.
func (s *Sync) commitCodeRequest(req *codeRequest) error {
// Write the node content to the membatch
s.membatch.codes[req.hash] = req.data
delete(s.codeReqs, req.hash)
s.fetches[len(req.path)]--
// Check all parents for completion
for _, parent := range req.parents {
parent.deps--
if parent.deps == 0 {
if err := s.commitNodeRequest(parent); err != nil {
return err return err
} }
} }

346
trie/sync_test.go
View file

@ -88,6 +88,13 @@ func checkTrieConsistency(db *Database, root common.Hash) error {
return it.Error() return it.Error()
} }
// trieElement represents the element in the state trie(bytecode or trie node).
type trieElement struct {
path string
hash common.Hash
syncPath SyncPath
}
// Tests that an empty trie is not scheduled for syncing. // Tests that an empty trie is not scheduled for syncing.
func TestEmptySync(t *testing.T) { func TestEmptySync(t *testing.T) {
dbA := NewDatabase(memorydb.New()) dbA := NewDatabase(memorydb.New())
@ -97,8 +104,8 @@ func TestEmptySync(t *testing.T) {
for i, trie := range []*Trie{emptyA, emptyB} { for i, trie := range []*Trie{emptyA, emptyB} {
sync := NewSync(trie.Hash(), memorydb.New(), nil) sync := NewSync(trie.Hash(), memorydb.New(), nil)
if nodes, paths, codes := sync.Missing(1); len(nodes) != 0 || len(paths) != 0 || len(codes) != 0 { if paths, nodes, codes := sync.Missing(1); len(paths) != 0 || len(nodes) != 0 || len(codes) != 0 {
t.Errorf("test %d: content requested for empty trie: %v, %v, %v", i, nodes, paths, codes) t.Errorf("test %d: content requested for empty trie: %v, %v, %v", i, paths, nodes, codes)
} }
} }
} }
@ -119,35 +126,38 @@ func testIterativeSync(t *testing.T, count int, bypath bool) {
triedb := NewDatabase(diskdb) triedb := NewDatabase(diskdb)
sched := NewSync(srcTrie.Hash(), diskdb, nil) sched := NewSync(srcTrie.Hash(), diskdb, nil)
nodes, paths, codes := sched.Missing(count) // The code requests are ignored here since there is no code
var ( // at the testing trie.
hashQueue []common.Hash paths, nodes, _ := sched.Missing(count)
pathQueue []SyncPath var elements []trieElement
) for i := 0; i < len(paths); i++ {
if !bypath { elements = append(elements, trieElement{
hashQueue = append(append(hashQueue[:0], nodes...), codes...) path: paths[i],
} else { hash: nodes[i],
hashQueue = append(hashQueue[:0], codes...) syncPath: NewSyncPath([]byte(paths[i])),
pathQueue = append(pathQueue[:0], paths...) })
} }
for len(hashQueue)+len(pathQueue) > 0 { for len(elements) > 0 {
results := make([]SyncResult, len(hashQueue)+len(pathQueue)) results := make([]NodeSyncResult, len(elements))
for i, hash := range hashQueue { if !bypath {
data, err := srcDb.Node(hash) for i, element := range elements {
if err != nil { data, err := srcDb.Node(element.hash)
t.Fatalf("failed to retrieve node data for hash %x: %v", hash, err) if err != nil {
t.Fatalf("failed to retrieve node data for hash %x: %v", element.hash, err)
}
results[i] = NodeSyncResult{element.path, data}
} }
results[i] = SyncResult{hash, data} } else {
} for i, element := range elements {
for i, path := range pathQueue { data, _, err := srcTrie.TryGetNode(element.syncPath[len(element.syncPath)-1])
data, _, err := srcTrie.TryGetNode(path[0]) if err != nil {
if err != nil { t.Fatalf("failed to retrieve node data for path %x: %v", element.path, err)
t.Fatalf("failed to retrieve node data for path %x: %v", path, err) }
results[i] = NodeSyncResult{element.path, data}
} }
results[len(hashQueue)+i] = SyncResult{crypto.Keccak256Hash(data), data}
} }
for _, result := range results { for _, result := range results {
if err := sched.Process(result); err != nil { if err := sched.ProcessNode(result); err != nil {
t.Fatalf("failed to process result %v", err) t.Fatalf("failed to process result %v", err)
} }
} }
@ -157,12 +167,14 @@ func testIterativeSync(t *testing.T, count int, bypath bool) {
} }
batch.Write() batch.Write()
nodes, paths, codes = sched.Missing(count) paths, nodes, _ = sched.Missing(count)
if !bypath { elements = elements[:0]
hashQueue = append(append(hashQueue[:0], nodes...), codes...) for i := 0; i < len(paths); i++ {
} else { elements = append(elements, trieElement{
hashQueue = append(hashQueue[:0], codes...) path: paths[i],
pathQueue = append(pathQueue[:0], paths...) hash: nodes[i],
syncPath: NewSyncPath([]byte(paths[i])),
})
} }
} }
// Cross check that the two tries are in sync // Cross check that the two tries are in sync
@ -180,21 +192,29 @@ func TestIterativeDelayedSync(t *testing.T) {
triedb := NewDatabase(diskdb) triedb := NewDatabase(diskdb)
sched := NewSync(srcTrie.Hash(), diskdb, nil) sched := NewSync(srcTrie.Hash(), diskdb, nil)
nodes, _, codes := sched.Missing(10000) // The code requests are ignored here since there is no code
queue := append(append([]common.Hash{}, nodes...), codes...) // at the testing trie.
paths, nodes, _ := sched.Missing(10000)
for len(queue) > 0 { var elements []trieElement
for i := 0; i < len(paths); i++ {
elements = append(elements, trieElement{
path: paths[i],
hash: nodes[i],
syncPath: NewSyncPath([]byte(paths[i])),
})
}
for len(elements) > 0 {
// Sync only half of the scheduled nodes // Sync only half of the scheduled nodes
results := make([]SyncResult, len(queue)/2+1) results := make([]NodeSyncResult, len(elements)/2+1)
for i, hash := range queue[:len(results)] { for i, element := range elements[:len(results)] {
data, err := srcDb.Node(hash) data, err := srcDb.Node(element.hash)
if err != nil { if err != nil {
t.Fatalf("failed to retrieve Node data for %x: %v", hash, err) t.Fatalf("failed to retrieve node data for %x: %v", element.hash, err)
} }
results[i] = SyncResult{hash, data} results[i] = NodeSyncResult{element.path, data}
} }
for _, result := range results { for _, result := range results {
if err := sched.Process(result); err != nil { if err := sched.ProcessNode(result); err != nil {
t.Fatalf("failed to process result %v", err) t.Fatalf("failed to process result %v", err)
} }
} }
@ -204,8 +224,15 @@ func TestIterativeDelayedSync(t *testing.T) {
} }
batch.Write() batch.Write()
nodes, _, codes = sched.Missing(10000) paths, nodes, _ = sched.Missing(10000)
queue = append(append(queue[len(results):], nodes...), codes...) elements = elements[len(results):]
for i := 0; i < len(paths); i++ {
elements = append(elements, trieElement{
path: paths[i],
hash: nodes[i],
syncPath: NewSyncPath([]byte(paths[i])),
})
}
} }
// Cross check that the two tries are in sync // Cross check that the two tries are in sync
checkTrieContents(t, triedb, srcTrie.Hash().Bytes(), srcData) checkTrieContents(t, triedb, srcTrie.Hash().Bytes(), srcData)
@ -226,24 +253,30 @@ func testIterativeRandomSync(t *testing.T, count int) {
triedb := NewDatabase(diskdb) triedb := NewDatabase(diskdb)
sched := NewSync(srcTrie.Hash(), diskdb, nil) sched := NewSync(srcTrie.Hash(), diskdb, nil)
queue := make(map[common.Hash]struct{}) // The code requests are ignored here since there is no code
nodes, _, codes := sched.Missing(count) // at the testing trie.
for _, hash := range append(nodes, codes...) { paths, nodes, _ := sched.Missing(count)
queue[hash] = struct{}{} queue := make(map[string]trieElement)
for i, path := range paths {
queue[path] = trieElement{
path: paths[i],
hash: nodes[i],
syncPath: NewSyncPath([]byte(paths[i])),
}
} }
for len(queue) > 0 { for len(queue) > 0 {
// Fetch all the queued nodes in a random order // Fetch all the queued nodes in a random order
results := make([]SyncResult, 0, len(queue)) results := make([]NodeSyncResult, 0, len(queue))
for hash := range queue { for path, element := range queue {
data, err := srcDb.Node(hash) data, err := srcDb.Node(element.hash)
if err != nil { if err != nil {
t.Fatalf("failed to retrieve Node data for %x: %v", hash, err) t.Fatalf("failed to retrieve node data for %x: %v", element.hash, err)
} }
results = append(results, SyncResult{hash, data}) results = append(results, NodeSyncResult{path, data})
} }
// Feed the retrieved results back and queue new tasks // Feed the retrieved results back and queue new tasks
for _, result := range results { for _, result := range results {
if err := sched.Process(result); err != nil { if err := sched.ProcessNode(result); err != nil {
t.Fatalf("failed to process result %v", err) t.Fatalf("failed to process result %v", err)
} }
} }
@ -253,10 +286,14 @@ func testIterativeRandomSync(t *testing.T, count int) {
} }
batch.Write() batch.Write()
queue = make(map[common.Hash]struct{}) paths, nodes, _ = sched.Missing(count)
nodes, _, codes = sched.Missing(count) queue = make(map[string]trieElement)
for _, hash := range append(nodes, codes...) { for i, path := range paths {
queue[hash] = struct{}{} queue[path] = trieElement{
path: path,
hash: nodes[i],
syncPath: NewSyncPath([]byte(path)),
}
} }
} }
// Cross check that the two tries are in sync // Cross check that the two tries are in sync
@ -274,20 +311,26 @@ func TestIterativeRandomDelayedSync(t *testing.T) {
triedb := NewDatabase(diskdb) triedb := NewDatabase(diskdb)
sched := NewSync(srcTrie.Hash(), diskdb, nil) sched := NewSync(srcTrie.Hash(), diskdb, nil)
queue := make(map[common.Hash]struct{}) // The code requests are ignored here since there is no code
nodes, _, codes := sched.Missing(10000) // at the testing trie.
for _, hash := range append(nodes, codes...) { paths, nodes, _ := sched.Missing(10000)
queue[hash] = struct{}{} queue := make(map[string]trieElement)
for i, path := range paths {
queue[path] = trieElement{
path: path,
hash: nodes[i],
syncPath: NewSyncPath([]byte(path)),
}
} }
for len(queue) > 0 { for len(queue) > 0 {
// Sync only half of the scheduled nodes, even those in random order // Sync only half of the scheduled nodes, even those in random order
results := make([]SyncResult, 0, len(queue)/2+1) results := make([]NodeSyncResult, 0, len(queue)/2+1)
for hash := range queue { for path, element := range queue {
data, err := srcDb.Node(hash) data, err := srcDb.Node(element.hash)
if err != nil { if err != nil {
t.Fatalf("failed to retrieve Node data for %x: %v", hash, err) t.Fatalf("failed to retrieve node data for %x: %v", element.hash, err)
} }
results = append(results, SyncResult{hash, data}) results = append(results, NodeSyncResult{path, data})
if len(results) >= cap(results) { if len(results) >= cap(results) {
break break
@ -295,7 +338,7 @@ func TestIterativeRandomDelayedSync(t *testing.T) {
} }
// Feed the retrieved results back and queue new tasks // Feed the retrieved results back and queue new tasks
for _, result := range results { for _, result := range results {
if err := sched.Process(result); err != nil { if err := sched.ProcessNode(result); err != nil {
t.Fatalf("failed to process result %v", err) t.Fatalf("failed to process result %v", err)
} }
} }
@ -305,11 +348,15 @@ func TestIterativeRandomDelayedSync(t *testing.T) {
} }
batch.Write() batch.Write()
for _, result := range results { for _, result := range results {
delete(queue, result.Hash) delete(queue, result.Path)
} }
nodes, _, codes = sched.Missing(10000) paths, nodes, _ = sched.Missing(10000)
for _, hash := range append(nodes, codes...) { for i, path := range paths {
queue[hash] = struct{}{} queue[path] = trieElement{
path: path,
hash: nodes[i],
syncPath: NewSyncPath([]byte(path)),
}
} }
} }
// Cross check that the two tries are in sync // Cross check that the two tries are in sync
@ -327,26 +374,35 @@ func TestDuplicateAvoidanceSync(t *testing.T) {
triedb := NewDatabase(diskdb) triedb := NewDatabase(diskdb)
sched := NewSync(srcTrie.Hash(), diskdb, nil) sched := NewSync(srcTrie.Hash(), diskdb, nil)
nodes, _, codes := sched.Missing(0) // The code requests are ignored here since there is no code
queue := append(append([]common.Hash{}, nodes...), codes...) // at the testing trie.
paths, nodes, _ := sched.Missing(0)
var elements []trieElement
for i := 0; i < len(paths); i++ {
elements = append(elements, trieElement{
path: paths[i],
hash: nodes[i],
syncPath: NewSyncPath([]byte(paths[i])),
})
}
requested := make(map[common.Hash]struct{}) requested := make(map[common.Hash]struct{})
for len(queue) > 0 { for len(elements) > 0 {
results := make([]SyncResult, len(queue)) results := make([]NodeSyncResult, len(elements))
for i, hash := range queue { for i, element := range elements {
data, err := srcDb.Node(hash) data, err := srcDb.Node(element.hash)
if err != nil { if err != nil {
t.Fatalf("failed to retrieve Node data for %x: %v", hash, err) t.Fatalf("failed to retrieve node data for %x: %v", element.hash, err)
} }
if _, ok := requested[hash]; ok { if _, ok := requested[element.hash]; ok {
t.Errorf("hash %x already requested once", hash) t.Errorf("hash %x already requested once", element.hash)
} }
requested[hash] = struct{}{} requested[element.hash] = struct{}{}
results[i] = SyncResult{hash, data} results[i] = NodeSyncResult{element.path, data}
} }
for _, result := range results { for _, result := range results {
if err := sched.Process(result); err != nil { if err := sched.ProcessNode(result); err != nil {
t.Fatalf("failed to process result %v", err) t.Fatalf("failed to process result %v", err)
} }
} }
@ -356,8 +412,15 @@ func TestDuplicateAvoidanceSync(t *testing.T) {
} }
batch.Write() batch.Write()
nodes, _, codes = sched.Missing(0) paths, nodes, _ = sched.Missing(0)
queue = append(append(queue[:0], nodes...), codes...) elements = elements[:0]
for i := 0; i < len(paths); i++ {
elements = append(elements, trieElement{
path: paths[i],
hash: nodes[i],
syncPath: NewSyncPath([]byte(paths[i])),
})
}
} }
// Cross check that the two tries are in sync // Cross check that the two tries are in sync
checkTrieContents(t, triedb, srcTrie.Hash().Bytes(), srcData) checkTrieContents(t, triedb, srcTrie.Hash().Bytes(), srcData)
@ -374,23 +437,34 @@ func TestIncompleteSync(t *testing.T) {
triedb := NewDatabase(diskdb) triedb := NewDatabase(diskdb)
sched := NewSync(srcTrie.Hash(), diskdb, nil) sched := NewSync(srcTrie.Hash(), diskdb, nil)
var added []common.Hash // The code requests are ignored here since there is no code
// at the testing trie.
nodes, _, codes := sched.Missing(1) var (
queue := append(append([]common.Hash{}, nodes...), codes...) added []common.Hash
for len(queue) > 0 { elements []trieElement
root = srcTrie.Hash()
)
paths, nodes, _ := sched.Missing(1)
for i := 0; i < len(paths); i++ {
elements = append(elements, trieElement{
path: paths[i],
hash: nodes[i],
syncPath: NewSyncPath([]byte(paths[i])),
})
}
for len(elements) > 0 {
// Fetch a batch of trie nodes // Fetch a batch of trie nodes
results := make([]SyncResult, len(queue)) results := make([]NodeSyncResult, len(elements))
for i, hash := range queue { for i, element := range elements {
data, err := srcDb.Node(hash) data, err := srcDb.Node(element.hash)
if err != nil { if err != nil {
t.Fatalf("failed to retrieve node data for %x: %v", hash, err) t.Fatalf("failed to retrieve node data for %x: %v", element.hash, err)
} }
results[i] = SyncResult{hash, data} results[i] = NodeSyncResult{element.path, data}
} }
// Process each of the trie nodes // Process each of the trie nodes
for _, result := range results { for _, result := range results {
if err := sched.Process(result); err != nil { if err := sched.ProcessNode(result); err != nil {
t.Fatalf("failed to process result %v", err) t.Fatalf("failed to process result %v", err)
} }
} }
@ -399,27 +473,36 @@ func TestIncompleteSync(t *testing.T) {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write() batch.Write()
for _, result := range results { for _, result := range results {
added = append(added, result.Hash) hash := crypto.Keccak256Hash(result.Data)
if hash != root {
added = append(added, hash)
}
// Check that all known sub-tries in the synced trie are complete // Check that all known sub-tries in the synced trie are complete
if err := checkTrieConsistency(triedb, result.Hash); err != nil { if err := checkTrieConsistency(triedb, hash); err != nil {
t.Fatalf("trie inconsistent: %v", err) t.Fatalf("trie inconsistent: %v", err)
} }
} }
// Fetch the next batch to retrieve // Fetch the next batch to retrieve
nodes, _, codes = sched.Missing(1) paths, nodes, _ = sched.Missing(1)
queue = append(append(queue[:0], nodes...), codes...) elements = elements[:0]
for i := 0; i < len(paths); i++ {
elements = append(elements, trieElement{
path: paths[i],
hash: nodes[i],
syncPath: NewSyncPath([]byte(paths[i])),
})
}
} }
// Sanity check that removing any node from the database is detected // Sanity check that removing any node from the database is detected
for _, node := range added[1:] { for _, hash := range added {
key := node.Bytes() value, _ := diskdb.Get(hash.Bytes())
value, _ := diskdb.Get(key) diskdb.Delete(hash.Bytes())
if err := checkTrieConsistency(triedb, root); err == nil {
diskdb.Delete(key) t.Fatalf("trie inconsistency not caught, missing: %x", hash)
if err := checkTrieConsistency(triedb, added[0]); err == nil {
t.Fatalf("trie inconsistency not caught, missing: %x", key)
} }
diskdb.Put(key, value) diskdb.Put(hash.Bytes(), value)
} }
} }
@ -434,21 +517,33 @@ func TestSyncOrdering(t *testing.T) {
triedb := NewDatabase(diskdb) triedb := NewDatabase(diskdb)
sched := NewSync(srcTrie.Hash(), diskdb, nil) sched := NewSync(srcTrie.Hash(), diskdb, nil)
nodes, paths, _ := sched.Missing(1) // The code requests are ignored here since there is no code
queue := append([]common.Hash{}, nodes...) // at the testing trie.
reqs := append([]SyncPath{}, paths...) var (
reqs []SyncPath
elements []trieElement
)
paths, nodes, _ := sched.Missing(1)
for i := 0; i < len(paths); i++ {
elements = append(elements, trieElement{
path: paths[i],
hash: nodes[i],
syncPath: NewSyncPath([]byte(paths[i])),
})
reqs = append(reqs, NewSyncPath([]byte(paths[i])))
}
for len(queue) > 0 { for len(elements) > 0 {
results := make([]SyncResult, len(queue)) results := make([]NodeSyncResult, len(elements))
for i, hash := range queue { for i, element := range elements {
data, err := srcDb.Node(hash) data, err := srcDb.Node(element.hash)
if err != nil { if err != nil {
t.Fatalf("failed to retrieve node data for %x: %v", hash, err) t.Fatalf("failed to retrieve node data for %x: %v", element.hash, err)
} }
results[i] = SyncResult{hash, data} results[i] = NodeSyncResult{element.path, data}
} }
for _, result := range results { for _, result := range results {
if err := sched.Process(result); err != nil { if err := sched.ProcessNode(result); err != nil {
t.Fatalf("failed to process result %v", err) t.Fatalf("failed to process result %v", err)
} }
} }
@ -458,9 +553,16 @@ func TestSyncOrdering(t *testing.T) {
} }
batch.Write() batch.Write()
nodes, paths, _ = sched.Missing(1) paths, nodes, _ = sched.Missing(1)
queue = append(queue[:0], nodes...) elements = elements[:0]
reqs = append(reqs, paths...) for i := 0; i < len(paths); i++ {
elements = append(elements, trieElement{
path: paths[i],
hash: nodes[i],
syncPath: NewSyncPath([]byte(paths[i])),
})
reqs = append(reqs, NewSyncPath([]byte(paths[i])))
}
} }
// Cross check that the two tries are in sync // Cross check that the two tries are in sync
checkTrieContents(t, triedb, srcTrie.Hash().Bytes(), srcData) checkTrieContents(t, triedb, srcTrie.Hash().Bytes(), srcData)

8
trie/trie.go
View file

@ -33,18 +33,18 @@ import (
// LeafCallback is a callback type invoked when a trie operation reaches a leaf // LeafCallback is a callback type invoked when a trie operation reaches a leaf
// node. // node.
// //
// The paths is a path tuple identifying a particular trie node either in a single // The keys is a path tuple identifying a particular trie node either in a single
// trie (account) or a layered trie (account -> storage). Each path in the tuple // trie (account) or a layered trie (account -> storage). Each key in the tuple
// is in the raw format(32 bytes). // is in the raw format(32 bytes).
// //
// The hexpath is a composite hexary path identifying the trie node. All the key // The path is a composite hexary path identifying the trie node. All the key
// bytes are converted to the hexary nibbles and composited with the parent path // bytes are converted to the hexary nibbles and composited with the parent path
// if the trie node is in a layered trie. // if the trie node is in a layered trie.
// //
// It's used by state sync and commit to allow handling external references // It's used by state sync and commit to allow handling external references
// between account and storage tries. And also it's used in the state healing // between account and storage tries. And also it's used in the state healing
// for extracting the raw states(leaf nodes) with corresponding paths. // for extracting the raw states(leaf nodes) with corresponding paths.
type LeafCallback func(paths [][]byte, hexpath []byte, leaf []byte, parent common.Hash) error type LeafCallback func(keys [][]byte, path []byte, leaf []byte, parent common.Hash, parentPath []byte) error
// Trie is a Merkle Patricia Trie. // Trie is a Merkle Patricia Trie.
// The zero value is an empty trie with no database. // The zero value is an empty trie with no database.

2
trie/trie_test.go
View file

@ -552,7 +552,7 @@ func BenchmarkCommitAfterHash(b *testing.B) {
benchmarkCommitAfterHash(b, nil) benchmarkCommitAfterHash(b, nil)
}) })
var a types.StateAccount var a types.StateAccount
onleaf := func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash) error { onleaf := func(_ [][]byte, _ []byte, leaf []byte, parent common.Hash, parentPath []byte) error {
rlp.DecodeBytes(leaf, &a) rlp.DecodeBytes(leaf, &a)
return nil return nil
} }