From 3d7a161287fc3effcfe8c1631e55cd9a8a6b558b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Fri, 8 Mar 2019 13:12:35 +0200 Subject: [PATCH] core, eth, trie: implement fast sync from and to a pruned node --- core/blockchain.go | 12 ++- core/state/iterator_test.go | 10 +- core/state/sync.go | 2 +- core/state/sync_test.go | 123 ++++++++++++++++-------- eth/downloader/downloader_test.go | 2 +- eth/downloader/statesync.go | 96 +++++++++++++------ trie/database.go | 105 ++++++++++++++------ trie/iterator_test.go | 41 ++++---- trie/secure_trie_test.go | 5 +- trie/sync.go | 153 ++++++++++++++++++++---------- trie/sync_test.go | 79 ++++++++------- trie/trie_test.go | 14 +-- 12 files changed, 418 insertions(+), 224 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index cc6ffa9398..fe647e681f 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -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 } diff --git a/core/state/iterator_test.go b/core/state/iterator_test.go index 69f51c4c7d..c548c610e0 100644 --- a/core/state/iterator_test.go +++ b/core/state/iterator_test.go @@ -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() } +*/ diff --git a/core/state/sync.go b/core/state/sync.go index 5d2666af4a..6401c7d7d7 100644 --- a/core/state/sync.go +++ b/core/state/sync.go @@ -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 } diff --git a/core/state/sync_test.go b/core/state/sync_test.go index ab4718b04c..4e912728d9 100644 --- a/core/state/sync_test.go +++ b/core/state/sync_test.go @@ -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: diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index 405d52a710..df40431212 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -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]) diff --git a/eth/downloader/statesync.go b/eth/downloader/statesync.go index f294600b37..3b73df48ff 100644 --- a/eth/downloader/statesync.go +++ b/eth/downloader/statesync.go @@ -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 diff --git a/trie/database.go b/trie/database.go index 3951f94176..d82bdbf0a6 100644 --- a/trie/database.go +++ b/trie/database.go @@ -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)) } diff --git a/trie/iterator_test.go b/trie/iterator_test.go index 88b8103fb3..cccb473ebf 100644 --- a/trie/iterator_test.go +++ b/trie/iterator_test.go @@ -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 { diff --git a/trie/secure_trie_test.go b/trie/secure_trie_test.go index f0ca6c8002..6b85735e37 100644 --- a/trie/secure_trie_test.go +++ b/trie/secure_trie_test.go @@ -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 diff --git a/trie/sync.go b/trie/sync.go index ac1f2b29cc..b8631f9694 100644 --- a/trie/sync.go +++ b/trie/sync.go @@ -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 { diff --git a/trie/sync_test.go b/trie/sync_test.go index d80070f3e8..2f137807af 100644 --- a/trie/sync_test.go +++ b/trie/sync_test.go @@ -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 { diff --git a/trie/trie_test.go b/trie/trie_test.go index cf133706fa..04504667ab 100644 --- a/trie/trie_test.go +++ b/trie/trie_test.go @@ -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 {