|
|
@ -34,14 +34,19 @@ var ErrNotRequested = errors.New("not requested") |
|
|
|
// node it already processed previously.
|
|
|
|
// node it already processed previously.
|
|
|
|
var ErrAlreadyProcessed = errors.New("already processed") |
|
|
|
var ErrAlreadyProcessed = errors.New("already processed") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// maxFetchesPerDepth is the maximum number of pending trie nodes per depth. The
|
|
|
|
|
|
|
|
// role of this value is to limit the number of trie nodes that get expanded in
|
|
|
|
|
|
|
|
// memory if the node was configured with a significant number of peers.
|
|
|
|
|
|
|
|
const maxFetchesPerDepth = 16384 |
|
|
|
|
|
|
|
|
|
|
|
// request represents a scheduled or already in-flight state retrieval request.
|
|
|
|
// request represents a scheduled or already in-flight state retrieval request.
|
|
|
|
type request struct { |
|
|
|
type request struct { |
|
|
|
|
|
|
|
path []byte // Merkle path leading to this node for prioritization
|
|
|
|
hash common.Hash // Hash of the node data content to retrieve
|
|
|
|
hash common.Hash // Hash of the node data content to retrieve
|
|
|
|
data []byte // Data content of the node, cached until all subtrees complete
|
|
|
|
data []byte // Data content of the node, cached until all subtrees complete
|
|
|
|
code bool // Whether this is a code entry
|
|
|
|
code bool // Whether this is a code entry
|
|
|
|
|
|
|
|
|
|
|
|
parents []*request // Parent state nodes referencing this entry (notify all upon completion)
|
|
|
|
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
|
|
|
|
|
|
|
|
deps int // Number of dependencies before allowed to commit this node
|
|
|
|
deps int // Number of dependencies before allowed to commit this node
|
|
|
|
|
|
|
|
|
|
|
|
callback LeafCallback // Callback to invoke if a leaf node it reached on this branch
|
|
|
|
callback LeafCallback // Callback to invoke if a leaf node it reached on this branch
|
|
|
@ -89,6 +94,7 @@ type Sync struct { |
|
|
|
nodeReqs map[common.Hash]*request // Pending requests pertaining to a trie node hash
|
|
|
|
nodeReqs map[common.Hash]*request // Pending requests pertaining to a trie node hash
|
|
|
|
codeReqs map[common.Hash]*request // Pending requests pertaining to a code hash
|
|
|
|
codeReqs map[common.Hash]*request // Pending requests pertaining to a code hash
|
|
|
|
queue *prque.Prque // Priority queue with the pending requests
|
|
|
|
queue *prque.Prque // Priority queue with the pending requests
|
|
|
|
|
|
|
|
fetches map[int]int // Number of active fetches per trie node depth
|
|
|
|
bloom *SyncBloom // Bloom filter for fast state existence checks
|
|
|
|
bloom *SyncBloom // Bloom filter for fast state existence checks
|
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
@ -100,14 +106,15 @@ func NewSync(root common.Hash, database ethdb.KeyValueReader, callback LeafCallb |
|
|
|
nodeReqs: make(map[common.Hash]*request), |
|
|
|
nodeReqs: make(map[common.Hash]*request), |
|
|
|
codeReqs: make(map[common.Hash]*request), |
|
|
|
codeReqs: make(map[common.Hash]*request), |
|
|
|
queue: prque.New(nil), |
|
|
|
queue: prque.New(nil), |
|
|
|
|
|
|
|
fetches: make(map[int]int), |
|
|
|
bloom: bloom, |
|
|
|
bloom: bloom, |
|
|
|
} |
|
|
|
} |
|
|
|
ts.AddSubTrie(root, 0, common.Hash{}, callback) |
|
|
|
ts.AddSubTrie(root, nil, common.Hash{}, callback) |
|
|
|
return ts |
|
|
|
return ts |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// AddSubTrie registers a new trie to the sync code, rooted at the designated parent.
|
|
|
|
// AddSubTrie registers a new trie to the sync code, rooted at the designated parent.
|
|
|
|
func (s *Sync) AddSubTrie(root common.Hash, depth int, parent common.Hash, callback LeafCallback) { |
|
|
|
func (s *Sync) AddSubTrie(root common.Hash, path []byte, parent common.Hash, callback LeafCallback) { |
|
|
|
// Short circuit if the trie is empty or already known
|
|
|
|
// Short circuit if the trie is empty or already known
|
|
|
|
if root == emptyRoot { |
|
|
|
if root == emptyRoot { |
|
|
|
return |
|
|
|
return |
|
|
@ -128,8 +135,8 @@ func (s *Sync) AddSubTrie(root common.Hash, depth int, parent common.Hash, callb |
|
|
|
} |
|
|
|
} |
|
|
|
// Assemble the new sub-trie sync request
|
|
|
|
// Assemble the new sub-trie sync request
|
|
|
|
req := &request{ |
|
|
|
req := &request{ |
|
|
|
|
|
|
|
path: path, |
|
|
|
hash: root, |
|
|
|
hash: root, |
|
|
|
depth: depth, |
|
|
|
|
|
|
|
callback: callback, |
|
|
|
callback: callback, |
|
|
|
} |
|
|
|
} |
|
|
|
// If this sub-trie has a designated parent, link them together
|
|
|
|
// If this sub-trie has a designated parent, link them together
|
|
|
@ -147,7 +154,7 @@ func (s *Sync) AddSubTrie(root common.Hash, depth int, parent common.Hash, callb |
|
|
|
// AddCodeEntry schedules the direct retrieval of a contract code that should not
|
|
|
|
// AddCodeEntry schedules the direct retrieval of a contract code that should not
|
|
|
|
// be interpreted as a trie node, but rather accepted and stored into the database
|
|
|
|
// be interpreted as a trie node, but rather accepted and stored into the database
|
|
|
|
// as is.
|
|
|
|
// as is.
|
|
|
|
func (s *Sync) AddCodeEntry(hash common.Hash, depth int, parent common.Hash) { |
|
|
|
func (s *Sync) AddCodeEntry(hash common.Hash, path []byte, parent common.Hash) { |
|
|
|
// Short circuit if the entry is empty or already known
|
|
|
|
// Short circuit if the entry is empty or already known
|
|
|
|
if hash == emptyState { |
|
|
|
if hash == emptyState { |
|
|
|
return |
|
|
|
return |
|
|
@ -170,9 +177,9 @@ func (s *Sync) AddCodeEntry(hash common.Hash, depth int, parent common.Hash) { |
|
|
|
} |
|
|
|
} |
|
|
|
// Assemble the new sub-trie sync request
|
|
|
|
// Assemble the new sub-trie sync request
|
|
|
|
req := &request{ |
|
|
|
req := &request{ |
|
|
|
|
|
|
|
path: path, |
|
|
|
hash: hash, |
|
|
|
hash: hash, |
|
|
|
code: true, |
|
|
|
code: true, |
|
|
|
depth: depth, |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
// If this sub-trie has a designated parent, link them together
|
|
|
|
// If this sub-trie has a designated parent, link them together
|
|
|
|
if parent != (common.Hash{}) { |
|
|
|
if parent != (common.Hash{}) { |
|
|
@ -190,7 +197,18 @@ func (s *Sync) AddCodeEntry(hash common.Hash, depth int, parent common.Hash) { |
|
|
|
func (s *Sync) Missing(max int) []common.Hash { |
|
|
|
func (s *Sync) Missing(max int) []common.Hash { |
|
|
|
var requests []common.Hash |
|
|
|
var requests []common.Hash |
|
|
|
for !s.queue.Empty() && (max == 0 || len(requests) < max) { |
|
|
|
for !s.queue.Empty() && (max == 0 || len(requests) < max) { |
|
|
|
requests = append(requests, s.queue.PopItem().(common.Hash)) |
|
|
|
// Retrieve th enext item in line
|
|
|
|
|
|
|
|
item, prio := s.queue.Peek() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// If we have too many already-pending tasks for this depth, throttle
|
|
|
|
|
|
|
|
depth := int(prio >> 56) |
|
|
|
|
|
|
|
if s.fetches[depth] > maxFetchesPerDepth { |
|
|
|
|
|
|
|
break |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
// Item is allowed to be scheduled, add it to the task list
|
|
|
|
|
|
|
|
s.queue.Pop() |
|
|
|
|
|
|
|
s.fetches[depth]++ |
|
|
|
|
|
|
|
requests = append(requests, item.(common.Hash)) |
|
|
|
} |
|
|
|
} |
|
|
|
return requests |
|
|
|
return requests |
|
|
|
} |
|
|
|
} |
|
|
@ -285,7 +303,11 @@ func (s *Sync) schedule(req *request) { |
|
|
|
// is a trie node and code has same hash. In this case two elements
|
|
|
|
// is a trie node and code has same hash. In this case two elements
|
|
|
|
// with same hash and same or different depth will be pushed. But it's
|
|
|
|
// with same hash and same or different depth will be pushed. But it's
|
|
|
|
// ok the worst case is the second response will be treated as duplicated.
|
|
|
|
// ok the worst case is the second response will be treated as duplicated.
|
|
|
|
s.queue.Push(req.hash, int64(req.depth)) |
|
|
|
prio := int64(len(req.path)) << 56 // depth >= 128 will never happen, storage leaves will be included in their parents
|
|
|
|
|
|
|
|
for i := 0; i < 14 && i < len(req.path); i++ { |
|
|
|
|
|
|
|
prio |= int64(15-req.path[i]) << (52 - i*4) // 15-nibble => lexicographic order
|
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
s.queue.Push(req.hash, prio) |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// children retrieves all the missing children of a state trie entry for future
|
|
|
|
// children retrieves all the missing children of a state trie entry for future
|
|
|
@ -293,8 +315,8 @@ func (s *Sync) schedule(req *request) { |
|
|
|
func (s *Sync) children(req *request, object node) ([]*request, error) { |
|
|
|
func (s *Sync) children(req *request, object node) ([]*request, error) { |
|
|
|
// Gather all the children of the node, irrelevant whether known or not
|
|
|
|
// Gather all the children of the node, irrelevant whether known or not
|
|
|
|
type child struct { |
|
|
|
type child struct { |
|
|
|
|
|
|
|
path []byte |
|
|
|
node node |
|
|
|
node node |
|
|
|
depth int |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
var children []child |
|
|
|
var children []child |
|
|
|
|
|
|
|
|
|
|
@ -302,14 +324,14 @@ func (s *Sync) children(req *request, object node) ([]*request, error) { |
|
|
|
case *shortNode: |
|
|
|
case *shortNode: |
|
|
|
children = []child{{ |
|
|
|
children = []child{{ |
|
|
|
node: node.Val, |
|
|
|
node: node.Val, |
|
|
|
depth: req.depth + len(node.Key), |
|
|
|
path: append(append([]byte(nil), req.path...), node.Key...), |
|
|
|
}} |
|
|
|
}} |
|
|
|
case *fullNode: |
|
|
|
case *fullNode: |
|
|
|
for i := 0; i < 17; i++ { |
|
|
|
for i := 0; i < 17; i++ { |
|
|
|
if node.Children[i] != nil { |
|
|
|
if node.Children[i] != nil { |
|
|
|
children = append(children, child{ |
|
|
|
children = append(children, child{ |
|
|
|
node: node.Children[i], |
|
|
|
node: node.Children[i], |
|
|
|
depth: req.depth + 1, |
|
|
|
path: append(append([]byte(nil), req.path...), byte(i)), |
|
|
|
}) |
|
|
|
}) |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
@ -322,7 +344,7 @@ func (s *Sync) children(req *request, object node) ([]*request, error) { |
|
|
|
// Notify any external watcher of a new key/value node
|
|
|
|
// Notify any external watcher of a new key/value node
|
|
|
|
if req.callback != nil { |
|
|
|
if req.callback != nil { |
|
|
|
if node, ok := (child.node).(valueNode); ok { |
|
|
|
if node, ok := (child.node).(valueNode); ok { |
|
|
|
if err := req.callback(node, req.hash); err != nil { |
|
|
|
if err := req.callback(req.path, node, req.hash); err != nil { |
|
|
|
return nil, err |
|
|
|
return nil, err |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
@ -346,9 +368,9 @@ func (s *Sync) children(req *request, object node) ([]*request, error) { |
|
|
|
} |
|
|
|
} |
|
|
|
// Locally unknown node, schedule for retrieval
|
|
|
|
// Locally unknown node, schedule for retrieval
|
|
|
|
requests = append(requests, &request{ |
|
|
|
requests = append(requests, &request{ |
|
|
|
|
|
|
|
path: child.path, |
|
|
|
hash: hash, |
|
|
|
hash: hash, |
|
|
|
parents: []*request{req}, |
|
|
|
parents: []*request{req}, |
|
|
|
depth: child.depth, |
|
|
|
|
|
|
|
callback: req.callback, |
|
|
|
callback: req.callback, |
|
|
|
}) |
|
|
|
}) |
|
|
|
} |
|
|
|
} |
|
|
@ -364,9 +386,11 @@ func (s *Sync) commit(req *request) (err error) { |
|
|
|
if req.code { |
|
|
|
if req.code { |
|
|
|
s.membatch.codes[req.hash] = req.data |
|
|
|
s.membatch.codes[req.hash] = req.data |
|
|
|
delete(s.codeReqs, req.hash) |
|
|
|
delete(s.codeReqs, req.hash) |
|
|
|
|
|
|
|
s.fetches[len(req.path)]-- |
|
|
|
} else { |
|
|
|
} else { |
|
|
|
s.membatch.nodes[req.hash] = req.data |
|
|
|
s.membatch.nodes[req.hash] = req.data |
|
|
|
delete(s.nodeReqs, req.hash) |
|
|
|
delete(s.nodeReqs, req.hash) |
|
|
|
|
|
|
|
s.fetches[len(req.path)]-- |
|
|
|
} |
|
|
|
} |
|
|
|
// Check all parents for completion
|
|
|
|
// Check all parents for completion
|
|
|
|
for _, parent := range req.parents { |
|
|
|
for _, parent := range req.parents { |
|
|
|