diff options
Diffstat (limited to 'trie/sync.go')
| -rw-r--r-- | trie/sync.go | 233 |
1 files changed, 233 insertions, 0 deletions
diff --git a/trie/sync.go b/trie/sync.go new file mode 100644 index 000000000..65cfd6ed8 --- /dev/null +++ b/trie/sync.go @@ -0,0 +1,233 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. + +package trie + +import ( + "fmt" + + "github.com/ethereum/go-ethereum/common" + "gopkg.in/karalabe/cookiejar.v2/collections/prque" +) + +// 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 + object *node // Target node to populate with retrieved data (hashnode originally) + + parents []*request // Parent state nodes referencing this entry (notify all upon completion) + depth int // Depth level within the trie the node is located to prioritize DFS + deps int // Number of dependencies before allowed to commit this node + + callback TrieSyncLeafCallback // 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. +type SyncResult struct { + Hash common.Hash // Hash of the originally unknown trie node + Data []byte // Data content of the retrieved node +} + +// TrieSyncLeafCallback is a callback type invoked when a trie sync reaches a +// leaf node. It's used by state syncing to check if the leaf node requires some +// further data syncing. +type TrieSyncLeafCallback func(leaf []byte, parent common.Hash) error + +// TrieSync is the main state trie synchronisation scheduler, which provides yet +// unknown trie hashes to retrieve, accepts node data associated with said hashes +// and reconstructs the trie steb by step until all is done. +type TrieSync struct { + database Database // State database for storing all the assembled node data + requests map[common.Hash]*request // Pending requests pertaining to a key hash + queue *prque.Prque // Priority queue with the pending requests +} + +// NewTrieSync creates a new trie data download scheduler. +func NewTrieSync(root common.Hash, database Database, callback TrieSyncLeafCallback) *TrieSync { + ts := &TrieSync{ + database: database, + requests: make(map[common.Hash]*request), + queue: prque.New(), + } + ts.AddSubTrie(root, 0, common.Hash{}, callback) + return ts +} + +// AddSubTrie registers a new trie to the sync code, rooted at the designated parent. +func (s *TrieSync) AddSubTrie(root common.Hash, depth int, parent common.Hash, callback TrieSyncLeafCallback) { + // Short circuit if the trie is empty + if root == emptyRoot { + return + } + // Assemble the new sub-trie sync request + node := node(hashNode(root.Bytes())) + req := &request{ + object: &node, + hash: root, + depth: depth, + callback: callback, + } + // If this sub-trie has a designated parent, link them together + if parent != (common.Hash{}) { + ancestor := s.requests[parent] + if ancestor == nil { + panic(fmt.Sprintf("sub-trie ancestor not found: %x", parent)) + } + ancestor.deps++ + req.parents = append(req.parents, ancestor) + } + s.schedule(req) +} + +// Missing retrieves the known missing nodes from the trie for retrieval. +func (s *TrieSync) Missing(max int) []common.Hash { + requests := []common.Hash{} + for !s.queue.Empty() && (max == 0 || len(requests) < max) { + requests = append(requests, s.queue.PopItem().(common.Hash)) + } + return requests +} + +// Process injects a batch of retrieved trie nodes data. +func (s *TrieSync) Process(results []SyncResult) (int, error) { + for i, item := range results { + // If the item was not requested, bail out + request := s.requests[item.Hash] + if request == nil { + return i, fmt.Errorf("not requested: %x", item.Hash) + } + // Decode the node data content and update the request + node, err := decodeNode(item.Data) + if err != nil { + return i, err + } + *request.object = node + request.data = item.Data + + // Create and schedule a request for all the children nodes + requests, err := s.children(request) + if err != nil { + return i, err + } + if len(requests) == 0 && request.deps == 0 { + s.commit(request) + continue + } + request.deps += len(requests) + for _, child := range requests { + s.schedule(child) + } + } + return 0, nil +} + +// schedule inserts a new state retrieval request into the fetch queue. If there +// is already a pending request for this node, the new request will be discarded +// and only a parent reference added to the old one. +func (s *TrieSync) schedule(req *request) { + // If we're already requesting this node, add a new reference and stop + if old, ok := s.requests[req.hash]; ok { + old.parents = append(old.parents, req.parents...) + return + } + // Schedule the request for future retrieval + s.queue.Push(req.hash, float32(req.depth)) + s.requests[req.hash] = req +} + +// children retrieves all the missing children of a state trie entry for future +// retrieval scheduling. +func (s *TrieSync) children(req *request) ([]*request, error) { + // Gather all the children of the node, irrelevant whether known or not + type child struct { + node *node + depth int + } + children := []child{} + + switch node := (*req.object).(type) { + case shortNode: + children = []child{{ + node: &node.Val, + depth: req.depth + len(node.Key), + }} + case fullNode: + for i := 0; i < 17; i++ { + if node[i] != nil { + children = append(children, child{ + node: &node[i], + depth: req.depth + 1, + }) + } + } + default: + panic(fmt.Sprintf("unknown node: %+v", node)) + } + // Iterate over the children, and request all unknown ones + 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(node, req.hash); err != nil { + return nil, err + } + } + } + // 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 + blob, _ := s.database.Get(node) + if local, err := decodeNode(blob); local != nil && err == nil { + *child.node = local + continue + } + // Locally unknown node, schedule for retrieval + requests = append(requests, &request{ + object: child.node, + hash: common.BytesToHash(node), + parents: []*request{req}, + depth: child.depth, + callback: req.callback, + }) + } + } + return requests, nil +} + +// commit finalizes a retrieval request and stores it into the database. If any +// of the referencing parent requests complete due to this commit, they are also +// committed themselves. +func (s *TrieSync) commit(req *request) error { + // Write the node content to disk + if err := s.database.Put(req.hash[:], req.data); err != nil { + return err + } + delete(s.requests, req.hash) + + // Check all parents for completion + for _, parent := range req.parents { + parent.deps-- + if parent.deps == 0 { + if err := s.commit(parent); err != nil { + return err + } + } + } + return nil +} |