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

core, eth, trie: implement fast sync from and to a pruned node

Browse source
This commit is contained in:
Péter Szilágyi 2019-03-08 13:12:35 +02:00
parent 2f469fd597
commit 3d7a161287
No known key found for this signature in database
GPG key ID: E9AE538CEDF8293D
12 changed files with 418 additions and 224 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
core/blockchain.go
View file

@ -215,8 +215,6 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
bc.gcsave = head.Root()
bc.stateCache.TrieDB().ForbidPrune(bc.gcsave)
}
bc.stateCache.TrieDB().ResumePruning()
// Take ownership of this particular state
go bc.update()
return bc, nil
@ -360,6 +358,16 @@ func (bc *BlockChain) FastSyncCommitHead(hash common.Hash) error {
bc.currentBlock.Store(block)
bc.chainmu.Unlock()
// Forbid pruning this chain segment until
triedb := bc.stateCache.TrieDB()
triedb.ForbidPrune(block.Root())
if bc.gcsave != (common.Hash{}) {
triedb.PermitPrune(bc.gcsave)
triedb.Dereference(bc.gcsave)
}
bc.gcsave = block.Root()
log.Info("Committed new head block", "number", block.Number(), "hash", hash)
return nil
}

10
core/state/iterator_test.go
View file

@ -16,14 +16,7 @@
package state
import (
"bytes"
"testing"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/ethdb"
)
/*
// Tests that the node iterator indeed walks over the entire database contents.
func TestNodeIteratorCoverage(t *testing.T) {
// Create some arbitrary test state to iterate
@ -63,3 +56,4 @@ func TestNodeIteratorCoverage(t *testing.T) {
}
it.Release()
}
*/

2
core/state/sync.go
View file

@ -33,7 +33,7 @@ func NewStateSync(root common.Hash, database ethdb.Reader) *trie.Sync {
if err := rlp.Decode(bytes.NewReader(leaf), &obj); err != nil {
return err
}
syncer.AddSubTrie(obj.Root, 64, parent, nil)
syncer.AddSubTrie(owner, obj.Root, 64, parent, nil)
syncer.AddRawEntry(common.BytesToHash(obj.CodeHash), 64, parent)
return nil
}

123
core/state/sync_test.go
View file

@ -96,7 +96,7 @@ func checkTrieConsistency(db ethdb.Database, root common.Hash) error {
if v, _ := db.Get(root[:]); v == nil {
return nil // Consider a non existent state consistent.
}
trie, err := trie.New(root, trie.NewDatabase(db))
trie, err := trie.New(root, trie.NewDatabase(db, false))
if err != nil {
return err
}
@ -143,15 +143,23 @@ func testIterativeStateSync(t *testing.T, batch int) {
dstDb := rawdb.NewMemoryDatabase()
sched := NewStateSync(srcRoot, dstDb)
queue := append([]common.Hash{}, sched.Missing(batch)...)
queue := append([]string{}, sched.Missing(batch)...)
for len(queue) > 0 {
results := make([]trie.SyncResult, len(queue))
for i, hash := range queue {
data, err := srcDb.TrieDB().Node(hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x", hash)
for i, key := range queue {
var (
data []byte
err error
)
if _, hash, code := trie.SplitNodeKey(key); !code {
data, err = srcDb.TrieDB().Node(hash)
} else {
data, err = srcDb.TrieDB().DiskDB().Get([]byte(key))
}
results[i] = trie.SyncResult{Hash: hash, Data: data}
if err != nil {
t.Fatalf("failed to retrieve node data for %x", key)
}
results[i] = trie.SyncResult{Key: key, Data: data}
}
if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err)
@ -175,16 +183,24 @@ func TestIterativeDelayedStateSync(t *testing.T) {
dstDb := rawdb.NewMemoryDatabase()
sched := NewStateSync(srcRoot, dstDb)
queue := append([]common.Hash{}, sched.Missing(0)...)
queue := append([]string{}, sched.Missing(0)...)
for len(queue) > 0 {
// Sync only half of the scheduled nodes
results := make([]trie.SyncResult, len(queue)/2+1)
for i, hash := range queue[:len(results)] {
data, err := srcDb.TrieDB().Node(hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x", hash)
for i, key := range queue[:len(results)] {
var (
data []byte
err error
)
if _, hash, code := trie.SplitNodeKey(key); !code {
data, err = srcDb.TrieDB().Node(hash)
} else {
data, err = srcDb.TrieDB().DiskDB().Get([]byte(key))
}
results[i] = trie.SyncResult{Hash: hash, Data: data}
if err != nil {
t.Fatalf("failed to retrieve node data for %x", key)
}
results[i] = trie.SyncResult{Key: key, Data: data}
}
if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err)
@ -212,19 +228,27 @@ func testIterativeRandomStateSync(t *testing.T, batch int) {
dstDb := rawdb.NewMemoryDatabase()
sched := NewStateSync(srcRoot, dstDb)
queue := make(map[common.Hash]struct{})
for _, hash := range sched.Missing(batch) {
queue[hash] = struct{}{}
queue := make(map[string]struct{})
for _, key := range sched.Missing(batch) {
queue[key] = struct{}{}
}
for len(queue) > 0 {
// Fetch all the queued nodes in a random order
results := make([]trie.SyncResult, 0, len(queue))
for hash := range queue {
data, err := srcDb.TrieDB().Node(hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x", hash)
for key := range queue {
var (
data []byte
err error
)
if _, hash, code := trie.SplitNodeKey(key); !code {
data, err = srcDb.TrieDB().Node(hash)
} else {
data, err = srcDb.TrieDB().DiskDB().Get([]byte(key))
}
results = append(results, trie.SyncResult{Hash: hash, Data: data})
if err != nil {
t.Fatalf("failed to retrieve node data for %x", key)
}
results = append(results, trie.SyncResult{Key: key, Data: data})
}
// Feed the retrieved results back and queue new tasks
if _, index, err := sched.Process(results); err != nil {
@ -233,9 +257,9 @@ func testIterativeRandomStateSync(t *testing.T, batch int) {
if index, err := sched.Commit(dstDb); err != nil {
t.Fatalf("failed to commit data #%d: %v", index, err)
}
queue = make(map[common.Hash]struct{})
for _, hash := range sched.Missing(batch) {
queue[hash] = struct{}{}
queue = make(map[string]struct{})
for _, key := range sched.Missing(batch) {
queue[key] = struct{}{}
}
}
// Cross check that the two states are in sync
@ -252,21 +276,29 @@ func TestIterativeRandomDelayedStateSync(t *testing.T) {
dstDb := rawdb.NewMemoryDatabase()
sched := NewStateSync(srcRoot, dstDb)
queue := make(map[common.Hash]struct{})
for _, hash := range sched.Missing(0) {
queue[hash] = struct{}{}
queue := make(map[string]struct{})
for _, key := range sched.Missing(0) {
queue[key] = struct{}{}
}
for len(queue) > 0 {
// Sync only half of the scheduled nodes, even those in random order
results := make([]trie.SyncResult, 0, len(queue)/2+1)
for hash := range queue {
delete(queue, hash)
for key := range queue {
delete(queue, key)
data, err := srcDb.TrieDB().Node(hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x", hash)
var (
data []byte
err error
)
if _, hash, code := trie.SplitNodeKey(key); !code {
data, err = srcDb.TrieDB().Node(hash)
} else {
data, err = srcDb.TrieDB().DiskDB().Get([]byte(key))
}
results = append(results, trie.SyncResult{Hash: hash, Data: data})
if err != nil {
t.Fatalf("failed to retrieve node data for %x", key)
}
results = append(results, trie.SyncResult{Key: key, Data: data})
if len(results) >= cap(results) {
break
@ -300,16 +332,24 @@ func TestIncompleteStateSync(t *testing.T) {
sched := NewStateSync(srcRoot, dstDb)
added := []common.Hash{}
queue := append([]common.Hash{}, sched.Missing(1)...)
queue := append([]string{}, sched.Missing(1)...)
for len(queue) > 0 {
// Fetch a batch of state nodes
results := make([]trie.SyncResult, len(queue))
for i, hash := range queue {
data, err := srcDb.TrieDB().Node(hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x", hash)
for i, key := range queue {
var (
data []byte
err error
)
if _, hash, code := trie.SplitNodeKey(key); !code {
data, err = srcDb.TrieDB().Node(hash)
} else {
data, err = srcDb.TrieDB().DiskDB().Get([]byte(key))
}
results[i] = trie.SyncResult{Hash: hash, Data: data}
if err != nil {
t.Fatalf("failed to retrieve node data for %x", key)
}
results[i] = trie.SyncResult{Key: key, Data: data}
}
// Process each of the state nodes
if _, index, err := sched.Process(results); err != nil {
@ -319,7 +359,10 @@ func TestIncompleteStateSync(t *testing.T) {
t.Fatalf("failed to commit data #%d: %v", index, err)
}
for _, result := range results {
added = append(added, result.Hash)
_, hash, code := trie.SplitNodeKey(result.Key)
if !code { // TODO(karlabe): we need this check back for code too!!
added = append(added, hash)
}
}
// Check that all known sub-tries added so far are complete or missing entirely.
checkSubtries:

2
eth/downloader/downloader_test.go
View file

@ -187,7 +187,7 @@ func (dl *downloadTester) CurrentFastBlock() *types.Block {
func (dl *downloadTester) FastSyncCommitHead(hash common.Hash) error {
// For now only check that the state trie is correct
if block := dl.GetBlockByHash(hash); block != nil {
_, err := trie.NewSecure(block.Root(), trie.NewDatabase(dl.stateDb), 0)
_, err := trie.NewSecure(block.Root(), trie.NewDatabase(dl.stateDb, false), 0)
return err
}
return fmt.Errorf("non existent block: %x", hash[:4])

96
eth/downloader/statesync.go
View file

@ -34,13 +34,13 @@ import (
// stateReq represents a batch of state fetch requests grouped together into
// a single data retrieval network packet.
type stateReq struct {
items []common.Hash // Hashes of the state items to download
tasks map[common.Hash]*stateTask // Download tasks to track previous attempts
timeout time.Duration // Maximum round trip time for this to complete
timer *time.Timer // Timer to fire when the RTT timeout expires
peer *peerConnection // Peer that we're requesting from
response [][]byte // Response data of the peer (nil for timeouts)
dropped bool // Flag whether the peer dropped off early
items []common.Hash // Hashes of the state items to download
tasks map[string]*stateTask // Download tasks to track previous attempts
timeout time.Duration // Maximum round trip time for this to complete
timer *time.Timer // Timer to fire when the RTT timeout expires
peer *peerConnection // Peer that we're requesting from
response [][]byte // Response data of the peer (nil for timeouts)
dropped bool // Flag whether the peer dropped off early
}
// timedOut returns if this request timed out.
@ -214,9 +214,11 @@ func (d *Downloader) runStateSync(s *stateSync) *stateSync {
type stateSync struct {
d *Downloader // Downloader instance to access and manage current peerset
sched *trie.Sync // State trie sync scheduler defining the tasks
keccak hash.Hash // Keccak256 hasher to verify deliveries with
tasks map[common.Hash]*stateTask // Set of tasks currently queued for retrieval
sched *trie.Sync // State trie sync scheduler defining the tasks
keccak hash.Hash // Keccak256 hasher to verify deliveries with
tasks map[string]*stateTask // Set of tasks currently queued for retrieval
taskmap map[common.Hash][]string // Set of tasks for the same hash (until eth/64)
numUncommitted int
bytesUncommitted int
@ -241,7 +243,8 @@ func newStateSync(d *Downloader, root common.Hash) *stateSync {
d: d,
sched: state.NewStateSync(root, d.stateDB),
keccak: sha3.NewLegacyKeccak256(),
tasks: make(map[common.Hash]*stateTask),
tasks: make(map[string]*stateTask),
taskmap: make(map[common.Hash][]string),
deliver: make(chan *stateReq),
cancel: make(chan struct{}),
done: make(chan struct{}),
@ -377,15 +380,18 @@ func (s *stateSync) assignTasks() {
func (s *stateSync) fillTasks(n int, req *stateReq) {
// Refill available tasks from the scheduler.
if len(s.tasks) < n {
new := s.sched.Missing(n - len(s.tasks))
for _, hash := range new {
s.tasks[hash] = &stateTask{make(map[string]struct{})}
keys := s.sched.Missing(n - len(s.tasks))
for _, key := range keys {
s.tasks[key] = &stateTask{make(map[string]struct{})}
_, hash, _ := trie.SplitNodeKey(key)
s.taskmap[hash] = append(s.taskmap[hash], key)
}
}
// Find tasks that haven't been tried with the request's peer.
req.items = make([]common.Hash, 0, n)
req.tasks = make(map[common.Hash]*stateTask, n)
for hash, t := range s.tasks {
req.tasks = make(map[string]*stateTask, n)
for key, t := range s.tasks {
// Stop when we've gathered enough requests
if len(req.items) == n {
break
@ -396,9 +402,12 @@ func (s *stateSync) fillTasks(n int, req *stateReq) {
}
// Assign the request to this peer
t.attempts[req.peer.id] = struct{}{}
_, hash, _ := trie.SplitNodeKey(key) // We don't care if it's account/storage node or code
req.items = append(req.items, hash)
req.tasks[hash] = t
delete(s.tasks, hash)
req.tasks[key] = t
delete(s.tasks, key)
}
}
@ -418,7 +427,7 @@ func (s *stateSync) process(req *stateReq) (int, error) {
// Iterate over all the delivered data and inject one-by-one into the trie
for _, blob := range req.response {
_, hash, err := s.processNodeData(blob)
_, hash, err := s.processNodeData(req.tasks, blob)
switch err {
case nil:
s.numUncommitted++
@ -431,13 +440,10 @@ func (s *stateSync) process(req *stateReq) (int, error) {
default:
return successful, fmt.Errorf("invalid state node %s: %v", hash.TerminalString(), err)
}
if _, ok := req.tasks[hash]; ok {
delete(req.tasks, hash)
}
}
// Put unfulfilled tasks back into the retry queue
npeers := s.d.peers.Len()
for hash, task := range req.tasks {
for key, task := range req.tasks {
// 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
// the node to retry the missing items (to avoid single-peer stalls).
@ -447,10 +453,11 @@ func (s *stateSync) process(req *stateReq) (int, error) {
// If we've requested the node too many times already, it may be a malicious
// sync where nobody has the right data. Abort.
if len(task.attempts) >= npeers {
_, hash, _ := trie.SplitNodeKey(key)
return successful, fmt.Errorf("state node %s failed with all peers (%d tries, %d peers)", hash.TerminalString(), len(task.attempts), npeers)
}
// Missing item, place into the retry queue.
s.tasks[hash] = task
s.tasks[key] = task
}
return successful, nil
}
@ -458,13 +465,44 @@ func (s *stateSync) process(req *stateReq) (int, error) {
// 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
// error occurred.
func (s *stateSync) processNodeData(blob []byte) (bool, 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(tasks map[string]*stateTask, blob []byte) (bool, common.Hash, error) {
// Calculate the hash of the returned node data
var hash common.Hash
s.keccak.Reset()
s.keccak.Write(blob)
s.keccak.Sum(res.Hash[:0])
committed, _, err := s.sched.Process([]trie.SyncResult{res})
return committed, res.Hash, err
s.keccak.Sum(hash[:0])
// Retrieve one key that matches the received data and that was also requested
// from this particular peer. If one is found, delete it from the taskmap.
var key string
keys := s.taskmap[hash]
for i, k := range keys {
if _, ok := tasks[k]; ok {
// Key found, remove it from the task sets
if len(keys) > 1 {
keys[i] = keys[len(keys)-1]
s.taskmap[hash] = keys[:len(keys)-1]
} else {
delete(s.taskmap, hash)
}
delete(tasks, k)
// Save the key to feed to the trie syncer
key = k
break
}
}
if key == "" {
return false, hash, trie.ErrNotRequested
}
committed, _, err := s.sched.Process([]trie.SyncResult{{Key: key, Data: blob}})
return committed, hash, err
}
// updateStats bumps the various state sync progress counters and displays a log

105
trie/database.go
View file

@ -17,6 +17,8 @@
package trie
import (
"bytes"
"errors"
"fmt"
"io"
"sync"
@ -92,7 +94,7 @@ func splitNodeKey(key string) (common.Hash, common.Hash) {
return common.BytesToHash([]byte(key[common.HashLength:])), common.BytesToHash([]byte(key[:common.HashLength]))
default:
panic(fmt.Sprintf("invalid node key: %s", key))
panic(fmt.Sprintf("invalid node key: %x", key))
}
}
@ -106,10 +108,11 @@ type Database struct {
pruner *pruner // Background pruner to remove unreferenced trie nodes
pruning uint32 // Flag whether the pruner is running (sanity checks)
cleans *bigcache.BigCache // GC friendly memory cache of clean node RLPs
dirties map[string]*cachedNode // Data and references relationships of dirty nodes
oldest string // Oldest tracked node, flush-list head
newest string // Newest tracked node, flush-list tail
cleans *bigcache.BigCache // GC friendly memory cache of clean node RLPs
dirties map[string]*cachedNode // Data and references relationships of dirty nodes
hashmap map[common.Hash][]*cachedNode // Hash to dirty mapping to cater for legacy fast sync (TODO(karalabe): remove!)
oldest string // Oldest tracked node, flush-list head
newest string // Newest tracked node, flush-list tail
preimages map[common.Hash][]byte // Preimages of nodes from the secure trie
@ -369,6 +372,7 @@ func NewDatabaseWithCache(diskdb ethdb.KeyValueStore, cache int, prune bool) *Da
noprune: make(map[common.Hash]struct{}),
cleans: cleans,
dirties: map[string]*cachedNode{metaRoot: {}},
hashmap: make(map[common.Hash][]*cachedNode),
preimages: make(map[common.Hash][]byte),
}
if prune {
@ -468,6 +472,7 @@ func (db *Database) insert(owner common.Hash, hash common.Hash, blob []byte, nod
return nil
})
db.dirties[key] = entry
db.hashmap[hash] = append(db.hashmap[hash], entry)
// Update the flush-list endpoints
if db.oldest == metaRoot {
@ -539,34 +544,34 @@ func (db *Database) Node(hash common.Hash) ([]byte, error) {
return enc, nil
}
}
// TODO(karalabe): We need 2 new retrieval mechanisms:
// - We need to retrieve from the dirty cache, needs some data struct extension (no owner)
// - We need to retrieve from the database, needs prefix iteration support (just needs the interface ext)
//
// The code below is what's needed to work, just without the 'owner' being available
/*
// Retrieve the node from the dirty cache if available
key := makeNodeKey(owner, hash)
// Retrieve the node from the dirty cache if available
var dirty *cachedNode
db.lock.RLock()
dirty := db.dirties[key]
db.lock.RUnlock()
db.lock.RLock()
if dirties := db.hashmap[hash]; len(dirties) > 0 {
dirty = dirties[0] // any version will do, we just want the rlp
}
db.lock.RUnlock()
if dirty != nil {
return dirty.rlp(), nil
}
// Content unavailable in memory, attempt to retrieve from disk
enc, err := db.diskdb.Get([]byte(key))
if err == nil && enc != nil {
if dirty != nil {
return dirty.rlp(), nil
}
// Content unavailable in memory, attempt to retrieve from disk
it := db.diskdb.NewIteratorWithPrefix(hash[:])
defer it.Release()
if it.Next() {
if bytes.HasPrefix(it.Key(), hash[:]) {
blob := common.CopyBytes(it.Value())
if db.cleans != nil {
db.cleans.Set(string(hash[:]), enc)
db.cleans.Set(string(hash[:]), blob)
memcacheCleanMissMeter.Mark(1)
memcacheCleanWriteMeter.Mark(int64(len(enc)))
memcacheCleanWriteMeter.Mark(int64(len(blob)))
}
return blob, nil
}
return enc, err
*/
return nil, nil
}
return nil, errors.New("not found")
}
// preimage retrieves a cached trie node pre-image from memory. If it cannot be
@ -721,7 +726,7 @@ func (db *Database) dereference(childOwner common.Hash, childHash common.Hash, p
db.dirties[child.flushPrev].flushNext = child.flushNext
db.dirties[child.flushNext].flushPrev = child.flushPrev
}
// Dereference all children and delete the node
// Dereference all children
child.iterateRefs(path, func(path []byte, hash common.Hash) error {
db.dereference(childOwner, hash, childOwner, childHash, path, derefs)
return nil
@ -730,7 +735,23 @@ func (db *Database) dereference(childOwner common.Hash, childHash common.Hash, p
owner, hash := splitNodeKey(key)
db.dereference(owner, hash, childOwner, childHash, nil, derefs)
}
// Delete the dirty node and also remove it from the hash index
delete(db.dirties, childKey)
index := db.hashmap[childHash]
entries := len(index)
for i, dirty := range index {
if dirty == child {
if len(index) > 1 {
index[i] = index[entries-1]
db.hashmap[childHash] = index[:entries-1]
} else {
delete(db.hashmap, childHash)
}
break
}
}
db.dirtiesSize -= common.StorageSize(common.HashLength + int(child.size))
if childOwner == faultyOwner && childHash == faultyHash {
@ -825,8 +846,22 @@ func (db *Database) Cap(limit common.StorageSize) error {
db.preimagesSize = 0
}
for db.oldest != oldest {
// Delete the oldest node and also remove it from the hash index
node := db.dirties[db.oldest]
_, hash := splitNodeKey(db.oldest)
delete(db.dirties, db.oldest)
index := db.hashmap[hash]
entries := len(index)
for i, dirty := range index {
if dirty == node {
index[i] = index[entries-1]
db.hashmap[hash] = index[:entries-1]
break
}
}
db.oldest = node.flushNext
db.dirtiesSize -= common.StorageSize(common.HashLength + int(node.size))
@ -982,7 +1017,7 @@ func (db *Database) uncache(owner common.Hash, hash common.Hash) {
db.dirties[node.flushPrev].flushNext = node.flushNext
db.dirties[node.flushNext].flushPrev = node.flushPrev
}
// Uncache the node's subtries and remove the node itself too
// Uncache the node's subtries
node.iterateRefs(nil, func(path []byte, child common.Hash) error {
db.uncache(owner, child)
return nil
@ -990,7 +1025,19 @@ func (db *Database) uncache(owner common.Hash, hash common.Hash) {
for child := range node.children {
db.uncache(splitNodeKey(child))
}
// Delete the cleaned up node and also remove it from the hash index
delete(db.dirties, key)
index := db.hashmap[hash]
entries := len(index)
for i, dirty := range index {
if dirty == node {
index[i] = index[entries-1]
db.hashmap[hash] = index[:entries-1]
break
}
}
db.dirtiesSize -= common.StorageSize(common.HashLength + int(node.size))
}

41
trie/iterator_test.go
View file

@ -113,8 +113,8 @@ func TestNodeIteratorCoverage(t *testing.T) {
t.Errorf("failed to retrieve reported node %x: %v", hash, err)
}
}
for hash, obj := range db.dirties {
if obj != nil && hash != (common.Hash{}) {
for key, obj := range db.dirties {
if _, hash := splitNodeKey(key); obj != nil && hash != (common.Hash{}) {
if _, ok := hashes[hash]; !ok {
t.Errorf("state entry not reported %x", hash)
}
@ -293,7 +293,7 @@ func TestIteratorContinueAfterErrorMemonly(t *testing.T) { testIteratorContinueA
func testIteratorContinueAfterError(t *testing.T, memonly bool) {
diskdb := memorydb.New()
triedb := NewDatabase(diskdb)
triedb := NewDatabase(diskdb, false)
tr, _ := New(common.Hash{}, triedb)
for _, val := range testdata1 {
@ -307,7 +307,7 @@ func testIteratorContinueAfterError(t *testing.T, memonly bool) {
var (
diskKeys [][]byte
memKeys []common.Hash
memKeys []string
)
if memonly {
memKeys = triedb.Nodes()
@ -325,7 +325,7 @@ func testIteratorContinueAfterError(t *testing.T, memonly bool) {
// Remove a random node from the database. It can't be the root node
// because that one is already loaded.
var (
rkey common.Hash
rkey string
rval []byte
robj *cachedNode
)
@ -333,9 +333,9 @@ func testIteratorContinueAfterError(t *testing.T, memonly bool) {
if memonly {
rkey = memKeys[rand.Intn(len(memKeys))]
} else {
copy(rkey[:], diskKeys[rand.Intn(len(diskKeys))])
rkey = string(diskKeys[rand.Intn(len(diskKeys))])
}
if rkey != tr.Hash() {
if rkey != makeNodeKey(common.Hash{}, tr.Hash()) {
break
}
}
@ -343,23 +343,25 @@ func testIteratorContinueAfterError(t *testing.T, memonly bool) {
robj = triedb.dirties[rkey]
delete(triedb.dirties, rkey)
} else {
rval, _ = diskdb.Get(rkey[:])
diskdb.Delete(rkey[:])
rval, _ = diskdb.Get([]byte(rkey))
diskdb.Delete([]byte(rkey))
}
// Iterate until the error is hit.
seen := make(map[string]bool)
it := tr.NodeIterator(nil)
checkIteratorNoDups(t, it, seen)
missing, ok := it.Error().(*MissingNodeError)
if !ok || missing.NodeHash != rkey {
_, hash := splitNodeKey(rkey)
if !ok || missing.NodeHash != hash {
t.Fatal("didn't hit missing node, got", it.Error())
}
// Add the node back and continue iteration.
if memonly {
triedb.dirties[rkey] = robj
} else {
diskdb.Put(rkey[:], rval)
diskdb.Put([]byte(rkey), rval)
}
checkIteratorNoDups(t, it, seen)
if it.Error() != nil {
@ -384,7 +386,7 @@ func TestIteratorContinueAfterSeekErrorMemonly(t *testing.T) {
func testIteratorContinueAfterSeekError(t *testing.T, memonly bool) {
// Commit test trie to db, then remove the node containing "bars".
diskdb := memorydb.New()
triedb := NewDatabase(diskdb)
triedb := NewDatabase(diskdb, false)
ctr, _ := New(common.Hash{}, triedb)
for _, val := range testdata1 {
@ -395,16 +397,17 @@ func testIteratorContinueAfterSeekError(t *testing.T, memonly bool) {
triedb.Commit(root, true)
}
barNodeHash := common.HexToHash("05041990364eb72fcb1127652ce40d8bab765f2bfe53225b1170d276cc101c2e")
barNodeKey := makeNodeKey(common.Hash{}, barNodeHash)
var (
barNodeBlob []byte
barNodeObj *cachedNode
)
if memonly {
barNodeObj = triedb.dirties[barNodeHash]
delete(triedb.dirties, barNodeHash)
barNodeObj = triedb.dirties[barNodeKey]
delete(triedb.dirties, barNodeKey)
} else {
barNodeBlob, _ = diskdb.Get(barNodeHash[:])
diskdb.Delete(barNodeHash[:])
barNodeBlob, _ = diskdb.Get([]byte(barNodeKey))
diskdb.Delete([]byte(barNodeKey))
}
// Create a new iterator that seeks to "bars". Seeking can't proceed because
// the node is missing.
@ -418,9 +421,9 @@ func testIteratorContinueAfterSeekError(t *testing.T, memonly bool) {
}
// Reinsert the missing node.
if memonly {
triedb.dirties[barNodeHash] = barNodeObj
triedb.dirties[barNodeKey] = barNodeObj
} else {
diskdb.Put(barNodeHash[:], barNodeBlob)
diskdb.Put([]byte(barNodeKey), barNodeBlob)
}
// Check that iteration produces the right set of values.
if err := checkIteratorOrder(testdata1[2:], NewIterator(it)); err != nil {

5
trie/secure_trie_test.go
View file

@ -28,15 +28,14 @@ import (
)
func newEmptySecure() *SecureTrie {
trie, _ := NewSecure(common.Hash{}, NewDatabase(memorydb.New()), 0)
trie, _ := NewSecure(common.Hash{}, NewDatabase(memorydb.New(), false), 0)
return trie
}
// makeTestSecureTrie creates a large enough secure trie for testing.
func makeTestSecureTrie() (*Database, *SecureTrie, map[string][]byte) {
// Create an empty trie
triedb := NewDatabase(memorydb.New())
triedb := NewDatabase(memorydb.New(), false)
trie, _ := NewSecure(common.Hash{}, triedb, 0)
// Fill it with some arbitrary data

153
trie/sync.go
View file

@ -33,11 +33,15 @@ var ErrNotRequested = errors.New("not requested")
// node it already processed previously.
var ErrAlreadyProcessed = errors.New("already processed")
// codePrefix is the database key prefix used to store raw trie entries (code).
var codePrefix = []byte("c")
// request represents a scheduled or already in-flight state retrieval request.
type request struct {
hash common.Hash // Hash of the node data content to retrieve
data []byte // Data content of the node, cached until all subtrees complete
raw bool // Whether this is a raw entry (code) or a trie node
key string // Key of the node data content to retrieve
path []byte // Merkle-Patricia path to track sub-trie ownership
data []byte // Data content of the node, cached until all subtrees complete
raw bool // Whether this is a raw entry (code) or a trie node
parents []*request // Parent state nodes referencing this entry (notify all upon completion)
depth int // Depth level within the trie the node is located to prioritise DFS
@ -46,25 +50,42 @@ type request struct {
callback LeafCallback // Callback to invoke if a leaf node it reached on this branch
}
// SyncResult is a simple list to return missing nodes along with their request
// hashes.
// SplitNodeKey interprets the specified key, splitting it into an owner:hash
// tuple, also specifying whether the key represents a bytecode.
func SplitNodeKey(key string) (common.Hash, common.Hash, bool) {
// Figure out if this key represents a byte code or not
var code bool
if len(key)%2 == 1 {
if key[0] != codePrefix[0] {
panic(fmt.Sprintf("invalid node prefix: %x", key[0]))
}
code, key = true, key[1:]
}
// Split the key into an [owner]:hash tuple and return
owner, hash := splitNodeKey(key)
return owner, hash, code
}
// SyncResult is a simple struct to return missing nodes along with their request
// keys.
type SyncResult struct {
Hash common.Hash // Hash of the originally unknown trie node
Data []byte // Data content of the retrieved node
Key string // Key of the originally unknown trie node
Data []byte // Data content of the retrieved node
}
// syncMemBatch is an in-memory buffer of successfully downloaded but not yet
// persisted data items.
type syncMemBatch struct {
batch map[common.Hash][]byte // In-memory membatch of recently completed items
order []common.Hash // Order of completion to prevent out-of-order data loss
batch map[string][]byte // In-memory membatch of recently completed items
order []string // Order of completion to prevent out-of-order data loss
}
// newSyncMemBatch allocates a new memory-buffer for not-yet persisted trie nodes.
func newSyncMemBatch() *syncMemBatch {
return &syncMemBatch{
batch: make(map[common.Hash][]byte),
order: make([]common.Hash, 0, 256),
batch: make(map[string][]byte),
order: make([]string, 0, 256),
}
}
@ -72,10 +93,10 @@ func newSyncMemBatch() *syncMemBatch {
// unknown trie hashes to retrieve, accepts node data associated with said hashes
// and reconstructs the trie step by step until all is done.
type Sync struct {
database ethdb.Reader // Persistent database to check for existing entries
membatch *syncMemBatch // Memory buffer to avoid frequent database writes
requests map[common.Hash]*request // Pending requests pertaining to a key hash
queue *prque.Prque // Priority queue with the pending requests
database ethdb.Reader // Persistent database to check for existing entries
membatch *syncMemBatch // Memory buffer to avoid frequent database writes
requests map[string]*request // Pending requests pertaining to a key hash
queue *prque.Prque // Priority queue with the pending requests
}
// NewSync creates a new trie data download scheduler.
@ -83,36 +104,42 @@ func NewSync(root common.Hash, database ethdb.Reader, callback LeafCallback) *Sy
ts := &Sync{
database: database,
membatch: newSyncMemBatch(),
requests: make(map[common.Hash]*request),
requests: make(map[string]*request),
queue: prque.New(nil),
}
ts.AddSubTrie(root, 0, common.Hash{}, callback)
ts.AddSubTrie(common.Hash{}, root, 0, common.Hash{}, callback)
return ts
}
// AddSubTrie registers a new trie to the sync code, rooted at the designated parent.
func (s *Sync) AddSubTrie(root common.Hash, depth int, parent common.Hash, callback LeafCallback) {
// Short circuit if the trie is empty or already known
// AddSubTrie registers a new trie to the sync code, rooted at the designated
// parent for completion tracking.
//
// Note, the root has an owner field for tracking which account (hash/path) it
// belongs to whereas parent does not. The reason is that Ethereum only ever
// supports 2 layers of tries (account -> storage), so a sub-trie will never
// ever have a parent who's owner is not the nil hash.
func (s *Sync) AddSubTrie(owner, root common.Hash, depth int, parent common.Hash, callback LeafCallback) {
// Short circuit if the trie is empty or already known _, hash := splitNodeKey(root)
if root == emptyRoot {
return
}
if _, ok := s.membatch.batch[root]; ok {
key := makeNodeKey(owner, root)
if _, ok := s.membatch.batch[key]; ok {
return
}
key := root.Bytes()
blob, _ := s.database.Get(key)
if local, err := decodeNode(key, blob, 0); local != nil && err == nil {
blob, _ := s.database.Get([]byte(key))
if local, err := decodeNode(root[:], blob, 0); local != nil && err == nil {
return
}
// Assemble the new sub-trie sync request
req := &request{
hash: root,
key: key,
depth: depth,
callback: callback,
}
// If this sub-trie has a designated parent, link them together
if parent != (common.Hash{}) {
ancestor := s.requests[parent]
if (parent != common.Hash{}) {
ancestor := s.requests[makeNodeKey(common.Hash{}, parent)]
if ancestor == nil {
panic(fmt.Sprintf("sub-trie ancestor not found: %x", parent))
}
@ -126,26 +153,35 @@ func (s *Sync) AddSubTrie(root common.Hash, depth int, parent common.Hash, callb
// interpreted as a trie node, but rather accepted and stored into the database
// as is. This method's goal is to support misc state metadata retrievals (e.g.
// contract code).
//
// Note, neither the hash, nor the parent has an owner specified. The reason is
// that in Ethereum, only bytecode is stored as a raw-entry referenced by the
// trie, but that is deduplicated to prevent attacks. The parent is always an
// account, so we know it's owner is the nil hash.
func (s *Sync) AddRawEntry(hash common.Hash, depth int, parent common.Hash) {
// Short circuit if the entry is empty or already known
if hash == emptyState {
return
}
if _, ok := s.membatch.batch[hash]; ok {
var (
keyRaw = append(codePrefix, hash[:]...)
keyStr = string(keyRaw)
)
if _, ok := s.membatch.batch[keyStr]; ok {
return
}
if ok, _ := s.database.Has(hash.Bytes()); ok {
if ok, _ := s.database.Has(keyRaw); ok {
return
}
// Assemble the new sub-trie sync request
req := &request{
hash: hash,
key: keyStr,
raw: true,
depth: depth,
}
// If this sub-trie has a designated parent, link them together
if parent != (common.Hash{}) {
ancestor := s.requests[parent]
if (parent != common.Hash{}) {
ancestor := s.requests[makeNodeKey(common.Hash{}, parent)]
if ancestor == nil {
panic(fmt.Sprintf("raw-entry ancestor not found: %x", parent))
}
@ -156,10 +192,18 @@ func (s *Sync) AddRawEntry(hash common.Hash, depth int, parent common.Hash) {
}
// Missing retrieves the known missing nodes from the trie for retrieval.
func (s *Sync) Missing(max int) []common.Hash {
var requests []common.Hash
//
// The returned strings can represent three different things:
// - `hash` if it's an account trie node
// - `owner + hash` if it's a storage trie node
// - `'c' + hash` if it's an account bydecode node
//
// Use trie.SplitNodeKey to get an accurate interpretation of what exactly a
// returned key means.
func (s *Sync) Missing(max int) []string {
var requests []string
for !s.queue.Empty() && (max == 0 || len(requests) < max) {
requests = append(requests, s.queue.PopItem().(common.Hash))
requests = append(requests, s.queue.PopItem().(string))
}
return requests
}
@ -172,7 +216,7 @@ func (s *Sync) Process(results []SyncResult) (bool, int, error) {
for i, item := range results {
// If the item was not requested, bail out
request := s.requests[item.Hash]
request := s.requests[item.Key]
if request == nil {
return committed, i, ErrNotRequested
}
@ -187,7 +231,8 @@ func (s *Sync) Process(results []SyncResult) (bool, int, error) {
continue
}
// Decode the node data content and update the request
node, err := decodeNode(item.Hash[:], item.Data, 0)
_, hash := splitNodeKey(item.Key)
node, err := decodeNode(hash[:], item.Data, 0)
if err != nil {
return committed, i, err
}
@ -216,7 +261,7 @@ func (s *Sync) Process(results []SyncResult) (bool, int, error) {
func (s *Sync) Commit(dbw ethdb.Writer) (int, error) {
// Dump the membatch into a database dbw
for i, key := range s.membatch.order {
if err := dbw.Put(key[:], s.membatch.batch[key]); err != nil {
if err := dbw.Put([]byte(key), s.membatch.batch[key]); err != nil {
return i, err
}
}
@ -237,13 +282,13 @@ func (s *Sync) Pending() int {
// and only a parent reference added to the old one.
func (s *Sync) schedule(req *request) {
// If we're already requesting this node, add a new reference and stop
if old, ok := s.requests[req.hash]; ok {
if old, ok := s.requests[req.key]; ok {
old.parents = append(old.parents, req.parents...)
return
}
// Schedule the request for future retrieval
s.queue.Push(req.hash, int64(req.depth))
s.requests[req.hash] = req
s.queue.Push(req.key, int64(req.depth))
s.requests[req.key] = req
}
// children retrieves all the missing children of a state trie entry for future
@ -251,6 +296,7 @@ func (s *Sync) schedule(req *request) {
func (s *Sync) children(req *request, object node) ([]*request, error) {
// Gather all the children of the node, irrelevant whether known or not
type child struct {
path []byte
node node
depth int
}
@ -259,6 +305,7 @@ func (s *Sync) children(req *request, object node) ([]*request, error) {
switch node := (object).(type) {
case *shortNode:
children = []child{{
path: append(common.CopyBytes(req.path), node.Key...),
node: node.Val,
depth: req.depth + len(node.Key),
}}
@ -266,6 +313,7 @@ func (s *Sync) children(req *request, object node) ([]*request, error) {
for i := 0; i < 17; i++ {
if node.Children[i] != nil {
children = append(children, child{
path: append(common.CopyBytes(req.path), byte(i)),
node: node.Children[i],
depth: req.depth + 1,
})
@ -275,12 +323,18 @@ func (s *Sync) children(req *request, object node) ([]*request, error) {
panic(fmt.Sprintf("unknown node: %+v", node))
}
// Iterate over the children, and request all unknown ones
owner, hash := splitNodeKey(req.key)
requests := make([]*request, 0, len(children))
for _, child := range children {
// Notify any external watcher of a new key/value node
if req.callback != nil {
if node, ok := (child.node).(valueNode); ok {
if err := req.callback(common.Hash{}, node, req.hash); err != nil {
if len(child.path) != 65 {
panic(fmt.Sprintf("invalid child path (len %d): %x", len(child.path), child.path))
}
owner := common.BytesToHash(hexToKeybytes(child.path[:64]))
if err := req.callback(owner, node, hash); err != nil {
return nil, err
}
}
@ -288,16 +342,17 @@ func (s *Sync) children(req *request, object node) ([]*request, error) {
// If the child references another node, resolve or schedule
if node, ok := (child.node).(hashNode); ok {
// Try to resolve the node from the local database
hash := common.BytesToHash(node)
if _, ok := s.membatch.batch[hash]; ok {
key := makeNodeKey(owner, common.BytesToHash(node))
if _, ok := s.membatch.batch[key]; ok {
continue
}
if ok, _ := s.database.Has(node); ok {
if ok, _ := s.database.Has([]byte(key)); ok {
continue
}
// Locally unknown node, schedule for retrieval
requests = append(requests, &request{
hash: hash,
key: key,
path: child.path,
parents: []*request{req},
depth: child.depth,
callback: req.callback,
@ -312,10 +367,10 @@ func (s *Sync) children(req *request, object node) ([]*request, error) {
// committed themselves.
func (s *Sync) commit(req *request) (err error) {
// Write the node content to the membatch
s.membatch.batch[req.hash] = req.data
s.membatch.order = append(s.membatch.order, req.hash)
s.membatch.batch[req.key] = req.data
s.membatch.order = append(s.membatch.order, req.key)
delete(s.requests, req.hash)
delete(s.requests, req.key)
// Check all parents for completion
for _, parent := range req.parents {

79
trie/sync_test.go
View file

@ -27,7 +27,7 @@ import (
// makeTestTrie create a sample test trie to test node-wise reconstruction.
func makeTestTrie() (*Database, *Trie, map[string][]byte) {
// Create an empty trie
triedb := NewDatabase(memorydb.New())
triedb := NewDatabase(memorydb.New(), false)
trie, _ := New(common.Hash{}, triedb)
// Fill it with some arbitrary data
@ -88,8 +88,8 @@ func checkTrieConsistency(db *Database, root common.Hash) error {
// Tests that an empty trie is not scheduled for syncing.
func TestEmptySync(t *testing.T) {
dbA := NewDatabase(memorydb.New())
dbB := NewDatabase(memorydb.New())
dbA := NewDatabase(memorydb.New(), false)
dbB := NewDatabase(memorydb.New(), false)
emptyA, _ := New(common.Hash{}, dbA)
emptyB, _ := New(emptyRoot, dbB)
@ -111,18 +111,19 @@ func testIterativeSync(t *testing.T, batch int) {
// Create a destination trie and sync with the scheduler
diskdb := memorydb.New()
triedb := NewDatabase(diskdb)
triedb := NewDatabase(diskdb, false)
sched := NewSync(srcTrie.Hash(), diskdb, nil)
queue := append([]common.Hash{}, sched.Missing(batch)...)
queue := append([]string{}, sched.Missing(batch)...)
for len(queue) > 0 {
results := make([]SyncResult, len(queue))
for i, hash := range queue {
for i, key := range queue {
_, hash, _ := SplitNodeKey(key)
data, err := srcDb.Node(hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x: %v", hash, err)
}
results[i] = SyncResult{hash, data}
results[i] = SyncResult{key, data}
}
if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err)
@ -144,19 +145,20 @@ func TestIterativeDelayedSync(t *testing.T) {
// Create a destination trie and sync with the scheduler
diskdb := memorydb.New()
triedb := NewDatabase(diskdb)
triedb := NewDatabase(diskdb, false)
sched := NewSync(srcTrie.Hash(), diskdb, nil)
queue := append([]common.Hash{}, sched.Missing(10000)...)
queue := append([]string{}, sched.Missing(10000)...)
for len(queue) > 0 {
// Sync only half of the scheduled nodes
results := make([]SyncResult, len(queue)/2+1)
for i, hash := range queue[:len(results)] {
for i, key := range queue[:len(results)] {
_, hash, _ := SplitNodeKey(key)
data, err := srcDb.Node(hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x: %v", hash, err)
}
results[i] = SyncResult{hash, data}
results[i] = SyncResult{key, data}
}
if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err)
@ -182,22 +184,23 @@ func testIterativeRandomSync(t *testing.T, batch int) {
// Create a destination trie and sync with the scheduler
diskdb := memorydb.New()
triedb := NewDatabase(diskdb)
triedb := NewDatabase(diskdb, false)
sched := NewSync(srcTrie.Hash(), diskdb, nil)
queue := make(map[common.Hash]struct{})
for _, hash := range sched.Missing(batch) {
queue[hash] = struct{}{}
queue := make(map[string]struct{})
for _, key := range sched.Missing(batch) {
queue[key] = struct{}{}
}
for len(queue) > 0 {
// Fetch all the queued nodes in a random order
results := make([]SyncResult, 0, len(queue))
for hash := range queue {
for key := range queue {
_, hash, _ := SplitNodeKey(key)
data, err := srcDb.Node(hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x: %v", hash, err)
}
results = append(results, SyncResult{hash, data})
results = append(results, SyncResult{key, data})
}
// Feed the retrieved results back and queue new tasks
if _, index, err := sched.Process(results); err != nil {
@ -206,9 +209,9 @@ func testIterativeRandomSync(t *testing.T, batch int) {
if index, err := sched.Commit(diskdb); err != nil {
t.Fatalf("failed to commit data #%d: %v", index, err)
}
queue = make(map[common.Hash]struct{})
for _, hash := range sched.Missing(batch) {
queue[hash] = struct{}{}
queue = make(map[string]struct{})
for _, key := range sched.Missing(batch) {
queue[key] = struct{}{}
}
}
// Cross check that the two tries are in sync
@ -223,22 +226,23 @@ func TestIterativeRandomDelayedSync(t *testing.T) {
// Create a destination trie and sync with the scheduler
diskdb := memorydb.New()
triedb := NewDatabase(diskdb)
triedb := NewDatabase(diskdb, false)
sched := NewSync(srcTrie.Hash(), diskdb, nil)
queue := make(map[common.Hash]struct{})
for _, hash := range sched.Missing(10000) {
queue[hash] = struct{}{}
queue := make(map[string]struct{})
for _, key := range sched.Missing(10000) {
queue[key] = struct{}{}
}
for len(queue) > 0 {
// Sync only half of the scheduled nodes, even those in random order
results := make([]SyncResult, 0, len(queue)/2+1)
for hash := range queue {
for key := range queue {
_, hash, _ := SplitNodeKey(key)
data, err := srcDb.Node(hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x: %v", hash, err)
}
results = append(results, SyncResult{hash, data})
results = append(results, SyncResult{key, data})
if len(results) >= cap(results) {
break
@ -252,7 +256,7 @@ func TestIterativeRandomDelayedSync(t *testing.T) {
t.Fatalf("failed to commit data #%d: %v", index, err)
}
for _, result := range results {
delete(queue, result.Hash)
delete(queue, result.Key)
}
for _, hash := range sched.Missing(10000) {
queue[hash] = struct{}{}
@ -270,15 +274,16 @@ func TestDuplicateAvoidanceSync(t *testing.T) {
// Create a destination trie and sync with the scheduler
diskdb := memorydb.New()
triedb := NewDatabase(diskdb)
triedb := NewDatabase(diskdb, false)
sched := NewSync(srcTrie.Hash(), diskdb, nil)
queue := append([]common.Hash{}, sched.Missing(0)...)
queue := append([]string{}, sched.Missing(0)...)
requested := make(map[common.Hash]struct{})
for len(queue) > 0 {
results := make([]SyncResult, len(queue))
for i, hash := range queue {
for i, key := range queue {
_, hash, _ := SplitNodeKey(key)
data, err := srcDb.Node(hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x: %v", hash, err)
@ -288,7 +293,7 @@ func TestDuplicateAvoidanceSync(t *testing.T) {
}
requested[hash] = struct{}{}
results[i] = SyncResult{hash, data}
results[i] = SyncResult{key, data}
}
if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err)
@ -310,20 +315,21 @@ func TestIncompleteSync(t *testing.T) {
// Create a destination trie and sync with the scheduler
diskdb := memorydb.New()
triedb := NewDatabase(diskdb)
triedb := NewDatabase(diskdb, false)
sched := NewSync(srcTrie.Hash(), diskdb, nil)
var added []common.Hash
queue := append([]common.Hash{}, sched.Missing(1)...)
queue := append([]string{}, sched.Missing(1)...)
for len(queue) > 0 {
// Fetch a batch of trie nodes
results := make([]SyncResult, len(queue))
for i, hash := range queue {
for i, key := range queue {
_, hash, _ := SplitNodeKey(key)
data, err := srcDb.Node(hash)
if err != nil {
t.Fatalf("failed to retrieve node data for %x: %v", hash, err)
}
results[i] = SyncResult{hash, data}
results[i] = SyncResult{key, data}
}
// Process each of the trie nodes
if _, index, err := sched.Process(results); err != nil {
@ -333,7 +339,8 @@ func TestIncompleteSync(t *testing.T) {
t.Fatalf("failed to commit data #%d: %v", index, err)
}
for _, result := range results {
added = append(added, result.Hash)
_, hash, _ := SplitNodeKey(result.Key)
added = append(added, hash)
}
// Check that all known sub-tries in the synced trie are complete
for _, root := range added {

14
trie/trie_test.go
View file

@ -45,7 +45,7 @@ func init() {
// Used for testing
func newEmpty() *Trie {
trie, _ := New(common.Hash{}, NewDatabase(memorydb.New()))
trie, _ := New(common.Hash{}, NewDatabase(memorydb.New(), false))
return trie
}
@ -69,7 +69,7 @@ func TestNull(t *testing.T) {
}
func TestMissingRoot(t *testing.T) {
trie, err := New(common.HexToHash("0beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33"), NewDatabase(memorydb.New()))
trie, err := New(common.HexToHash("0beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33"), NewDatabase(memorydb.New(), false))
if trie != nil {
t.Error("New returned non-nil trie for invalid root")
}
@ -83,7 +83,7 @@ func TestMissingNodeMemonly(t *testing.T) { testMissingNode(t, true) }
func testMissingNode(t *testing.T, memonly bool) {
diskdb := memorydb.New()
triedb := NewDatabase(diskdb)
triedb := NewDatabase(diskdb, false)
trie, _ := New(common.Hash{}, triedb)
updateString(trie, "120000", "qwerqwerqwerqwerqwerqwerqwerqwer")
@ -121,7 +121,7 @@ func testMissingNode(t *testing.T, memonly bool) {
hash := common.HexToHash("0xe1d943cc8f061a0c0b98162830b970395ac9315654824bf21b73b891365262f9")
if memonly {
delete(triedb.dirties, hash)
delete(triedb.dirties, makeNodeKey(common.Hash{}, hash))
} else {
diskdb.Delete(hash[:])
}
@ -345,7 +345,7 @@ func TestCacheUnload(t *testing.T) {
// The branch containing it is loaded from DB exactly two times:
// in the 0th and 6th iteration.
diskdb := &countingDB{KeyValueStore: trie.db.diskdb, gets: make(map[string]int)}
triedb := NewDatabase(diskdb)
triedb := NewDatabase(diskdb, false)
trie, _ = New(root, triedb)
trie.SetCacheLimit(5)
for i := 0; i < 12; i++ {
@ -414,7 +414,7 @@ func (randTest) Generate(r *rand.Rand, size int) reflect.Value {
}
func runRandTest(rt randTest) bool {
triedb := NewDatabase(memorydb.New())
triedb := NewDatabase(memorydb.New(), false)
tr, _ := New(common.Hash{}, triedb)
values := make(map[string]string) // tracks content of the trie
@ -602,7 +602,7 @@ func tempDB() (string, *Database) {
if err != nil {
panic(fmt.Sprintf("can't create temporary database: %v", err))
}
return dir, NewDatabase(diskdb)
return dir, NewDatabase(diskdb, false)
}
func getString(trie *Trie, k string) []byte {