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, ethdb, trie: historical state pruning

Browse source
This commit is contained in:
Péter Szilágyi 2018-06-22 15:37:43 +03:00
parent 2cfe0bed9f
commit 2f469fd597
No known key found for this signature in database
GPG key ID: E9AE538CEDF8293D
23 changed files with 1289 additions and 255 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

65
core/blockchain.go
View file

@ -101,9 +101,11 @@ type BlockChain struct {
chainConfig *params.ChainConfig // Chain & network configuration
cacheConfig *CacheConfig // Cache configuration for pruning
db ethdb.Database // Low level persistent database to store final content in
triegc *prque.Prque // Priority queue mapping block numbers to tries to gc
gcproc time.Duration // Accumulates canonical block processing for trie dumping
db ethdb.Database // Low level persistent database to store final content in
gcqueue *prque.Prque // Priority queue mapping block numbers to tries to gc
gcsave common.Hash // Root hash of the last trie committed to disk
gcproc time.Duration // Accumulates canonical block processing for trie dumping
hc *HeaderChain
rmLogsFeed event.Feed
@ -166,7 +168,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
chainConfig: chainConfig,
cacheConfig: cacheConfig,
db: db,
triegc: prque.New(nil),
gcqueue: prque.New(nil),
stateCache: state.NewDatabaseWithCache(db, cacheConfig.TrieCleanLimit),
quit: make(chan struct{}),
shouldPreserve: shouldPreserve,
@ -207,6 +209,14 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
}
}
}
// Forbid the genesis state (forever) and latest state (temporarilly) from being pruned
bc.stateCache.TrieDB().ForbidPrune(bc.genesisBlock.Root())
if head := bc.CurrentBlock(); head.NumberU64() > 0 {
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
@ -691,7 +701,15 @@ func (bc *BlockChain) GetUnclesInChain(block *types.Block, length int) []*types.
// TrieNode retrieves a blob of data associated with a trie node (or code hash)
// either from ephemeral in-memory cache, or from persistent storage.
func (bc *BlockChain) TrieNode(hash common.Hash) ([]byte, error) {
return bc.stateCache.TrieDB().Node(hash)
// Attempt to satisfy this request with a trie node
if blob, err := bc.stateCache.TrieDB().Node(hash); blob != nil && err == nil {
return blob, nil
}
// Trie node not found, it may be a bytecode
if blob := rawdb.ReadCode(bc.db, hash); blob != nil {
return blob, nil
}
return nil, errors.New("not found")
}
// Stop stops the blockchain service. If any imports are currently in progress
@ -707,6 +725,9 @@ func (bc *BlockChain) Stop() {
bc.wg.Wait()
// Terminate the pruner, we don't want it to remove recent stuff while quitting
bc.stateCache.TrieDB().TerminatePruning()
// Ensure the state of a recent block is also stored to disk before exiting.
// We're writing three different states to catch different restart scenarios:
// - HEAD: So we don't need to reprocess any blocks in the general case
@ -725,11 +746,11 @@ func (bc *BlockChain) Stop() {
}
}
}
for !bc.triegc.Empty() {
triedb.Dereference(bc.triegc.PopItem().(common.Hash))
for !bc.gcqueue.Empty() {
triedb.Dereference(bc.gcqueue.PopItem().(common.Hash))
}
if size, _ := triedb.Size(); size != 0 {
log.Error("Dangling trie nodes after full cleanup")
log.Error("Dangling trie nodes after full cleanup", "size", size)
}
}
log.Info("Blockchain manager stopped")
@ -960,21 +981,25 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
}
rawdb.WriteBlock(bc.db, block)
// Pause pruning while the tries are being mutated
triedb := bc.stateCache.TrieDB()
triedb.PausePruning()
defer triedb.ResumePruning()
// Commit the state into the dirty memory-cache and either flush, or garbage collect
root, err := state.Commit(bc.chainConfig.IsEIP158(block.Number()))
if err != nil {
return NonStatTy, err
}
triedb := bc.stateCache.TrieDB()
// If we're running an archive node, always flush
if bc.cacheConfig.Disabled {
if err := triedb.Commit(root, false); err != nil {
return NonStatTy, err
}
} else {
// Full but not archive node, do proper garbage collection
triedb.Reference(root, common.Hash{}) // metadata reference to keep trie alive
bc.triegc.Push(root, -int64(block.NumberU64()))
triedb.Reference(common.Hash{}, root, common.Hash{}) // metadata reference to keep trie alive
bc.gcqueue.Push(root, -int64(block.NumberU64()))
if current := block.NumberU64(); current > triesInMemory {
// If we exceeded our memory allowance, flush matured singleton nodes to disk
@ -1005,13 +1030,21 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
triedb.Commit(header.Root, true)
lastWrite = chosen
bc.gcproc = 0
// A new snapshot was flushed to disk, swap the prune allowance
triedb.ForbidPrune(header.Root)
if bc.gcsave != (common.Hash{}) {
triedb.PermitPrune(bc.gcsave)
triedb.Dereference(bc.gcsave)
}
bc.gcsave = header.Root
}
}
// Garbage collect anything below our required write retention
for !bc.triegc.Empty() {
root, number := bc.triegc.Pop()
for !bc.gcqueue.Empty() {
root, number := bc.gcqueue.Pop()
if uint64(-number) > chosen {
bc.triegc.Push(root, number)
bc.gcqueue.Push(root, number)
break
}
triedb.Dereference(root.(common.Hash))

1
core/genesis.go
View file

@ -260,6 +260,7 @@ func (g *Genesis) ToBlock(db ethdb.Database) *types.Block {
head.Difficulty = params.GenesisDifficulty
}
statedb.Commit(false)
statedb.Database().TrieDB().Reference(common.Hash{}, root, common.Hash{})
statedb.Database().TrieDB().Commit(root, true)
return types.NewBlock(head, nil, nil, nil)

43
core/rawdb/accessors_state.go Normal file
View file

@ -0,0 +1,43 @@
// Copyright 2019 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 rawdb
import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
)
// ReadCode retrieves the bytecode associated with a given hash.
func ReadCode(db ethdb.Reader, hash common.Hash) []byte {
code, _ := db.Get(codeKey(hash))
return code
}
// WriteCode stores the bytecode associated with a given hash.
func WriteCode(db ethdb.Writer, hash common.Hash, code []byte) {
if err := db.Put(codeKey(hash), code); err != nil {
log.Crit("Failed to store bytecode", "err", err)
}
}
// DeleteCode removes the bytecode associated with a given hash.
func DeleteCode(db ethdb.Deleter, hash common.Hash) {
if err := db.Delete(codeKey(hash)); err != nil {
log.Crit("Failed to delete bytecode", "err", err)
}
}

7
core/rawdb/schema.go
View file

@ -53,6 +53,8 @@ var (
txLookupPrefix = []byte("l") // txLookupPrefix + hash -> transaction/receipt lookup metadata
bloomBitsPrefix = []byte("B") // bloomBitsPrefix + bit (uint16 big endian) + section (uint64 big endian) + hash -> bloom bits
codePrefix = []byte("c") // codePrefix + hash -> bytecode
preimagePrefix = []byte("secure-key-") // preimagePrefix + hash -> preimage
configPrefix = []byte("ethereum-config-") // config prefix for the db
@ -128,6 +130,11 @@ func bloomBitsKey(bit uint, section uint64, hash common.Hash) []byte {
return key
}
// codeKey = codePrefix + hash
func codeKey(hash common.Hash) []byte {
return append(codePrefix, hash.Bytes()...)
}
// preimageKey = preimagePrefix + hash
func preimageKey(hash common.Hash) []byte {
return append(preimagePrefix, hash.Bytes()...)

12
core/state/database.go
View file

@ -17,10 +17,12 @@
package state
import (
"errors"
"fmt"
"sync"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/trie"
lru "github.com/hashicorp/golang-lru"
@ -84,7 +86,7 @@ func NewDatabase(db ethdb.Database) Database {
func NewDatabaseWithCache(db ethdb.Database, cache int) Database {
csc, _ := lru.New(codeSizeCacheSize)
return &cachingDB{
db: trie.NewDatabaseWithCache(db, cache),
db: trie.NewDatabaseWithCache(db, cache, true),
codeSizeCache: csc,
}
}
@ -127,7 +129,7 @@ func (db *cachingDB) pushTrie(t *trie.SecureTrie) {
// OpenStorageTrie opens the storage trie of an account.
func (db *cachingDB) OpenStorageTrie(addrHash, root common.Hash) (Trie, error) {
return trie.NewSecure(root, db.db, 0)
return trie.NewSecureWithOwner(addrHash, root, db.db, 0)
}
// CopyTrie returns an independent copy of the given trie.
@ -144,11 +146,11 @@ func (db *cachingDB) CopyTrie(t Trie) Trie {
// ContractCode retrieves a particular contract's code.
func (db *cachingDB) ContractCode(addrHash, codeHash common.Hash) ([]byte, error) {
code, err := db.db.Node(codeHash)
if err == nil {
if code := rawdb.ReadCode(db.db.DiskDB().(ethdb.Database), codeHash); code != nil {
db.codeSizeCache.Add(codeHash, len(code))
return code, nil
}
return code, err
return nil, errors.New("not found")
}
// ContractCodeSize retrieves a particular contracts code's size.

13
core/state/statedb.go
View file

@ -24,8 +24,10 @@ import (
"sort"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/trie"
@ -635,7 +637,8 @@ func (s *StateDB) Commit(deleteEmptyObjects bool) (root common.Hash, err error)
case isDirty:
// Write any contract code associated with the state object
if stateObject.code != nil && stateObject.dirtyCode {
s.db.TrieDB().InsertBlob(common.BytesToHash(stateObject.CodeHash()), stateObject.code)
rawdb.WriteCode(s.db.TrieDB().DiskDB().(ethdb.Database), common.BytesToHash(stateObject.CodeHash()), stateObject.code)
//s.db.TrieDB().DiskDB().(ethdb.Database).Put(stateObject.CodeHash(), stateObject.code)
stateObject.dirtyCode = false
}
// Write any storage changes in the state object to its storage trie.
@ -648,17 +651,13 @@ func (s *StateDB) Commit(deleteEmptyObjects bool) (root common.Hash, err error)
delete(s.stateObjectsDirty, addr)
}
// Write trie changes.
root, err = s.trie.Commit(func(leaf []byte, parent common.Hash) error {
root, err = s.trie.Commit(func(owner common.Hash, leaf []byte, parent common.Hash) error {
var account Account
if err := rlp.DecodeBytes(leaf, &account); err != nil {
return nil
}
if account.Root != emptyState {
s.db.TrieDB().Reference(account.Root, parent)
}
code := common.BytesToHash(account.CodeHash)
if code != emptyCode {
s.db.TrieDB().Reference(code, parent)
s.db.TrieDB().Reference(owner, account.Root, parent)
}
return nil
})

2
core/state/sync.go
View file

@ -28,7 +28,7 @@ import (
// NewStateSync create a new state trie download scheduler.
func NewStateSync(root common.Hash, database ethdb.Reader) *trie.Sync {
var syncer *trie.Sync
callback := func(leaf []byte, parent common.Hash) error {
callback := func(owner common.Hash, leaf []byte, parent common.Hash) error {
var obj Account
if err := rlp.Decode(bytes.NewReader(leaf), &obj); err != nil {
return err

8
eth/api_tracer.go
View file

@ -304,10 +304,10 @@ func (api *PrivateDebugAPI) traceChain(ctx context.Context, start, end *types.Bl
failed = err
break
}
// Reference the trie twice, once for us, once for the tracer
database.TrieDB().Reference(root, common.Hash{})
// Reference the trie twice, once for us, once for the trancer
database.TrieDB().Reference(common.Hash{}, root, common.Hash{})
if number >= origin {
database.TrieDB().Reference(root, common.Hash{})
database.TrieDB().Reference(common.Hash{}, root, common.Hash{})
}
// Dereference all past tries we ourselves are done working with
if proot != (common.Hash{}) {
@ -688,7 +688,7 @@ func (api *PrivateDebugAPI) computeStateDB(block *types.Block, reexec uint64) (*
if err := statedb.Reset(root); err != nil {
return nil, fmt.Errorf("state reset after block %d failed: %v", block.NumberU64(), err)
}
database.TrieDB().Reference(root, common.Hash{})
database.TrieDB().Reference(common.Hash{}, root, common.Hash{})
if proot != (common.Hash{}) {
database.TrieDB().Dereference(proot)
}

6
les/handler.go
View file

@ -653,7 +653,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if err != nil {
continue
}
code, _ := statedb.Database().TrieDB().Node(common.BytesToHash(account.CodeHash))
code := rawdb.ReadCode(pm.chainDb, common.BytesToHash(account.CodeHash))
data = append(data, code)
if bytes += len(code); bytes >= softResponseLimit {
@ -938,7 +938,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
return errResp(ErrRequestRejected, "")
}
go func() {
trieDb := trie.NewDatabase(rawdb.NewTable(pm.chainDb, light.ChtTablePrefix))
trieDb := trie.NewDatabase(rawdb.NewTable(pm.chainDb, light.ChtTablePrefix), false)
for i, req := range req.Reqs {
if i != 0 && !task.waitOrStop() {
return
@ -1003,7 +1003,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
var prefix string
if root, prefix = pm.getHelperTrie(req.Type, req.TrieIdx); root != (common.Hash{}) {
auxTrie, _ = trie.New(root, trie.NewDatabase(rawdb.NewTable(pm.chainDb, prefix)))
auxTrie, _ = trie.New(root, trie.NewDatabase(rawdb.NewTable(pm.chainDb, prefix), false))
}
}
if req.AuxReq == auxRoot {

4
light/postprocess.go
View file

@ -159,7 +159,7 @@ func NewChtIndexer(db ethdb.Database, odr OdrBackend, size, confirms uint64) *co
diskdb: db,
odr: odr,
trieTable: trieTable,
triedb: trie.NewDatabaseWithCache(trieTable, 1), // Use a tiny cache only to keep memory down
triedb: trie.NewDatabaseWithCache(trieTable, 1, false), // Use a tiny cache only to keep memory down
sectionSize: size,
}
return core.NewChainIndexer(db, rawdb.NewTable(db, "chtIndex-"), backend, size, confirms, time.Millisecond*100, "cht")
@ -281,7 +281,7 @@ func NewBloomTrieIndexer(db ethdb.Database, odr OdrBackend, parentSize, size uin
diskdb: db,
odr: odr,
trieTable: trieTable,
triedb: trie.NewDatabaseWithCache(trieTable, 1), // Use a tiny cache only to keep memory down
triedb: trie.NewDatabaseWithCache(trieTable, 1, false), // Use a tiny cache only to keep memory down
parentSize: parentSize,
size: size,
}

4
light/trie.go
View file

@ -151,7 +151,7 @@ func (t *odrTrie) do(key []byte, fn func() error) error {
for {
var err error
if t.trie == nil {
t.trie, err = trie.New(t.id.Root, trie.NewDatabase(t.db.backend.Database()))
t.trie, err = trie.New(t.id.Root, trie.NewDatabase(t.db.backend.Database(), false))
}
if err == nil {
err = fn()
@ -177,7 +177,7 @@ func newNodeIterator(t *odrTrie, startkey []byte) trie.NodeIterator {
// Open the actual non-ODR trie if that hasn't happened yet.
if t.trie == nil {
it.do(func() error {
t, err := trie.New(t.id.Root, trie.NewDatabase(t.db.backend.Database()))
t, err := trie.New(t.id.Root, trie.NewDatabase(t.db.backend.Database(), false))
if err == nil {
it.t.trie = t
}

599
trie/database.go
View file

@ -20,10 +20,13 @@ import (
"fmt"
"io"
"sync"
"sync/atomic"
"time"
"github.com/allegro/bigcache"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
@ -52,27 +55,72 @@ var (
// secureKeyPrefix is the database key prefix used to store trie node preimages.
var secureKeyPrefix = []byte("secure-key-")
// secureKeyLength is the length of the above prefix + 32byte hash.
const secureKeyLength = 11 + 32
// metaRoot is the identifier of the global memcache root that anchors the block
// accounts tries for garbage collection.
const metaRoot = ""
// DatabaseReader wraps the Get and Has method of a backing store for the trie.
type DatabaseReader interface {
// Get retrieves the value associated with key from the database.
Get(key []byte) (value []byte, err error)
// Has retrieves whether a key is present in the database.
Has(key []byte) (bool, error)
}
// makeNodeKey returns the database key for a trie node.
func makeNodeKey(owner common.Hash, hash common.Hash) string {
if hash == (common.Hash{}) {
return metaRoot
}
if owner == (common.Hash{}) {
return string(hash[:])
}
return string(append(hash[:], owner[:]...))
}
// splitNodeKey returns the composing hashes of a trie node key.
func splitNodeKey(key string) (common.Hash, common.Hash) {
switch len(key) {
case 0:
return common.Hash{}, common.Hash{}
case common.HashLength:
return common.Hash{}, common.BytesToHash([]byte(key))
case 2 * common.HashLength:
return common.BytesToHash([]byte(key[common.HashLength:])), common.BytesToHash([]byte(key[:common.HashLength]))
default:
panic(fmt.Sprintf("invalid node key: %s", key))
}
}
// Database is an intermediate write layer between the trie data structures and
// the disk database. The aim is to accumulate trie writes in-memory and only
// periodically flush a couple tries to disk, garbage collecting the remainder.
type Database struct {
diskdb ethdb.KeyValueStore // Persistent storage for matured trie nodes
diskdb ethdb.KeyValueStore // Persistent storage for matured trie nodes
noprune map[common.Hash]struct{} // Root hashes of the tries that aren't prunable
cleans *bigcache.BigCache // GC friendly memory cache of clean node RLPs
dirties map[common.Hash]*cachedNode // Data and references relationships of dirty nodes
oldest common.Hash // Oldest tracked node, flush-list head
newest common.Hash // Newest tracked node, flush-list tail
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
preimages map[common.Hash][]byte // Preimages of nodes from the secure trie
seckeybuf [secureKeyLength]byte // Ephemeral buffer for calculating preimage keys
gctime time.Duration // Time spent on garbage collection since last commit
gcnodes uint64 // Nodes garbage collected since last commit
gcsize common.StorageSize // Data storage garbage collected since last commit
prunetime time.Duration // Time spent on disk pruning since last commit
prunenodes uint64 // Nodes pruned from disk since last commit
prunesize common.StorageSize // Data storage pruned from disk since last commit
flushtime time.Duration // Time spent on data flushing since last commit
flushnodes uint64 // Nodes flushed since last commit
flushsize common.StorageSize // Data storage flushed since last commit
@ -99,7 +147,6 @@ type rawFullNode [17]node
func (n rawFullNode) canUnload(uint16, uint16) bool { panic("this should never end up in a live trie") }
func (n rawFullNode) cache() (hashNode, bool) { panic("this should never end up in a live trie") }
func (n rawFullNode) fstring(ind string) string { panic("this should never end up in a live trie") }
func (n rawFullNode) EncodeRLP(w io.Writer) error {
var nodes [17]node
@ -114,6 +161,20 @@ func (n rawFullNode) EncodeRLP(w io.Writer) error {
return rlp.Encode(w, nodes)
}
func (n rawFullNode) String() string { return n.fstring("") }
func (n rawFullNode) fstring(ind string) string {
resp := fmt.Sprintf("[\n%s ", ind)
for i, node := range n {
if node == nil {
resp += fmt.Sprintf("%s: <nil> ", indices[i])
} else {
resp += fmt.Sprintf("%s: %v", indices[i], node.fstring(ind+" "))
}
}
return resp + fmt.Sprintf("\n%s] ", ind)
}
// rawShortNode represents only the useful data content of a short node, with the
// caches and flags stripped out to minimize its data storage. This type honors
// the same RLP encoding as the original parent.
@ -122,9 +183,20 @@ type rawShortNode struct {
Val node
}
func (n rawShortNode) canUnload(uint16, uint16) bool { panic("this should never end up in a live trie") }
func (n rawShortNode) cache() (hashNode, bool) { panic("this should never end up in a live trie") }
func (n rawShortNode) fstring(ind string) string { panic("this should never end up in a live trie") }
func (n *rawShortNode) canUnload(uint16, uint16) bool {
panic("this should never end up in a live trie")
}
func (n *rawShortNode) cache() (hashNode, bool) { panic("this should never end up in a live trie") }
func (n *rawShortNode) EncodeRLP(w io.Writer) error {
return rlp.Encode(w, &shortNode{Key: hexToCompact(n.Key), Val: n.Val})
}
func (n *rawShortNode) String() string { return n.fstring("") }
func (n *rawShortNode) fstring(ind string) string {
return fmt.Sprintf("{%x: %v} ", n.Key, n.Val.fstring(ind+" "))
}
// cachedNode is all the information we know about a single cached node in the
// memory database write layer.
@ -132,11 +204,11 @@ type cachedNode struct {
node node // Cached collapsed trie node, or raw rlp data
size uint16 // Byte size of the useful cached data
parents uint32 // Number of live nodes referencing this one
children map[common.Hash]uint16 // External children referenced by this node
parents uint32 // Number of live nodes referencing this one
children map[string]uint16 // External children referenced by this node
flushPrev common.Hash // Previous node in the flush-list
flushNext common.Hash // Next node in the flush-list
flushPrev string // Previous node in the flush-list
flushNext string // Next node in the flush-list
}
// rlp returns the raw rlp encoded blob of the cached node, either directly from
@ -161,34 +233,46 @@ func (n *cachedNode) obj(hash common.Hash, cachegen uint16) node {
return expandNode(hash[:], n.node, cachegen)
}
// childs returns all the tracked children of this node, both the implicit ones
// from inside the node as well as the explicit ones from outside the node.
func (n *cachedNode) childs() []common.Hash {
children := make([]common.Hash, 0, 16)
for child := range n.children {
children = append(children, child)
// iterateRefs walks the embedded children of the cached node, tracking the
// internal path and invoking the provided callback on all hash nodes.
func (n *cachedNode) iterateRefs(path []byte, onHashNode func([]byte, common.Hash) error) error {
if _, ok := n.node.(rawNode); ok {
return nil
}
if _, ok := n.node.(rawNode); !ok {
gatherChildren(n.node, &children)
}
return children
return iterateRefs(n.node, path, onHashNode)
}
// gatherChildren traverses the node hierarchy of a collapsed storage node and
// retrieves all the hashnode children.
func gatherChildren(n node, children *[]common.Hash) {
// iterateRefs traverses the node hierarchy of a cached node and invokes the
// provided callback on all hash nodes.
func iterateRefs(n node, path []byte, onHashNode func([]byte, common.Hash) error) error {
switch n := n.(type) {
case *rawShortNode:
gatherChildren(n.Val, children)
return iterateRefs(n.Val, append(path, n.Key...), onHashNode)
case *shortNode:
return iterateRefs(n.Val, append(path, n.Key...), onHashNode)
case rawFullNode:
for i := 0; i < 16; i++ {
gatherChildren(n[i], children)
if err := iterateRefs(n[i], append(path, byte(i)), onHashNode); err != nil {
return err
}
}
return nil
case *fullNode:
for i := 0; i < 16; i++ {
if err := iterateRefs(n.Children[i], append(path, byte(i)), onHashNode); err != nil {
return err
}
}
return nil
case hashNode:
*children = append(*children, common.BytesToHash(n))
return onHashNode(path, common.BytesToHash(n))
case valueNode, nil:
return nil
default:
panic(fmt.Sprintf("unknown node type: %T", n))
@ -201,7 +285,7 @@ func simplifyNode(n node) node {
switch n := n.(type) {
case *shortNode:
// Short nodes discard the flags and cascade
return &rawShortNode{Key: n.Key, Val: simplifyNode(n.Val)}
return &rawShortNode{Key: compactToHex(n.Key), Val: simplifyNode(n.Val)}
case *fullNode:
// Full nodes discard the flags and cascade
@ -228,7 +312,7 @@ func expandNode(hash hashNode, n node, cachegen uint16) node {
case *rawShortNode:
// Short nodes need key and child expansion
return &shortNode{
Key: compactToHex(n.Key),
Key: n.Key,
Val: expandNode(nil, n.Val, cachegen),
flags: nodeFlag{
hash: hash,
@ -262,14 +346,14 @@ func expandNode(hash hashNode, n node, cachegen uint16) node {
// NewDatabase creates a new trie database to store ephemeral trie content before
// its written out to disk or garbage collected. No read cache is created, so all
// data retrievals will hit the underlying disk database.
func NewDatabase(diskdb ethdb.KeyValueStore) *Database {
return NewDatabaseWithCache(diskdb, 0)
func NewDatabase(diskdb ethdb.KeyValueStore, prune bool) *Database {
return NewDatabaseWithCache(diskdb, 0, prune)
}
// NewDatabaseWithCache creates a new trie database to store ephemeral trie content
// before its written out to disk or garbage collected. It also acts as a read cache
// for nodes loaded from disk.
func NewDatabaseWithCache(diskdb ethdb.KeyValueStore, cache int) *Database {
func NewDatabaseWithCache(diskdb ethdb.KeyValueStore, cache int, prune bool) *Database {
var cleans *bigcache.BigCache
if cache > 0 {
cleans, _ = bigcache.NewBigCache(bigcache.Config{
@ -280,59 +364,122 @@ func NewDatabaseWithCache(diskdb ethdb.KeyValueStore, cache int) *Database {
HardMaxCacheSize: cache,
})
}
return &Database{
db := &Database{
diskdb: diskdb,
noprune: make(map[common.Hash]struct{}),
cleans: cleans,
dirties: map[common.Hash]*cachedNode{{}: {}},
dirties: map[string]*cachedNode{metaRoot: {}},
preimages: make(map[common.Hash][]byte),
}
if prune {
db.pruner = newPruner(db)
}
return db
}
// ResumePruning permits the pruner to continue deleting unreferenced trie nodes.
// It is essential to only ever resume pruning after all data is commited, capped
// and properly referenced, otherwise the pruner might delete data that's *going-
// to-be* referenced.
func (db *Database) ResumePruning() {
if db.pruner != nil {
atomic.StoreUint32(&db.pruning, 1)
db.pruner.resume()
}
}
// DiskDB retrieves the persistent storage backing the trie database.
func (db *Database) DiskDB() ethdb.Reader {
return db.diskdb
// PausePruning waits until the pruner is done with processing its current task
// and then pauses it so a new trie might be properly integrated into the dirty
// caches and reference counts.
func (db *Database) PausePruning() {
if db.pruner != nil {
atomic.StoreUint32(&db.pruning, 0)
db.pruner.pause()
}
}
// InsertBlob writes a new reference tracked blob to the memory database if it's
// yet unknown. This method should only be used for non-trie nodes that require
// reference counting, since trie nodes are garbage collected directly through
// their embedded children.
func (db *Database) InsertBlob(hash common.Hash, blob []byte) {
// TerminatePruning waits until the pruner is done with processing all its queued
// tasls and then permanently terminates it.
func (db *Database) TerminatePruning() {
if db.pruner != nil {
atomic.StoreUint32(&db.pruning, 0)
db.pruner.terminate()
db.pruner = nil // TODO(karalabe): raceyyyy....
}
}
// ForbidPrune adds a root hash to the list of tries that are disallowed from
// being pruned. The only two ever to be used are the genesis trie and the last
// committed trie (snapshot).
func (db *Database) ForbidPrune(root common.Hash) {
db.lock.Lock()
defer db.lock.Unlock()
db.insert(hash, blob, rawNode(blob))
log.Debug("Forbidding pruner to delete trie", "root", root)
db.noprune[root] = struct{}{}
}
// PermitPrune allows a particular trie to be pruned from the disk database. To
// actually run the pruner, please call a dereference on the root hash.
func (db *Database) PermitPrune(root common.Hash) {
db.lock.Lock()
defer db.lock.Unlock()
log.Debug("Permitting pruner to delete trie", "root", root)
delete(db.noprune, root)
}
// DiskDB retrieves the persistent storage backing the trie database.
func (db *Database) DiskDB() DatabaseReader {
return db.diskdb
}
// insert inserts a collapsed trie node into the memory database. This method is
// a more generic version of InsertBlob, supporting both raw blob insertions as
// well ex trie node insertions. The blob must always be specified to allow proper
// size tracking.
func (db *Database) insert(hash common.Hash, blob []byte, node node) {
func (db *Database) insert(owner common.Hash, hash common.Hash, blob []byte, node node) {
if owner == faultyOwner && hash == faultyHash {
log.Error("Inserting sensitive dex trie node", "owner", owner, "hash", hash)
}
// If the node's already cached, skip
if _, ok := db.dirties[hash]; ok {
key := makeNodeKey(owner, hash)
if _, ok := db.dirties[key]; ok {
return
}
// If the node is already on disk, skip as well
/*if blob, err := db.diskdb.Get([]byte(key)); blob != nil && err == nil {
return
}*/
// Create the cached entry for this node
entry := &cachedNode{
node: simplifyNode(node),
size: uint16(len(blob)),
flushPrev: db.newest,
}
for _, child := range entry.childs() {
if c := db.dirties[child]; c != nil {
// Track all the implicit references (explicits must be empty)
entry.iterateRefs(nil, func(path []byte, child common.Hash) error {
if c := db.dirties[makeNodeKey(owner, child)]; c != nil {
c.parents++
if owner == faultyOwner && child == faultyHash {
log.Error("Inc. refcount on sensitive dex trie node", "parent", hexutil.Encode(entry.rlp()), "owner", owner, "hash", child, "parents", fmt.Sprintf("%d->%d", c.parents-1, c.parents))
}
}
}
db.dirties[hash] = entry
return nil
})
db.dirties[key] = entry
// Update the flush-list endpoints
if db.oldest == (common.Hash{}) {
db.oldest, db.newest = hash, hash
if db.oldest == metaRoot {
db.oldest, db.newest = key, key
} else {
db.dirties[db.newest].flushNext, db.newest = hash, hash
db.dirties[db.newest].flushNext, db.newest = key, key
}
db.dirtiesSize += common.StorageSize(common.HashLength + entry.size)
if owner == faultyOwner && hash == faultyHash {
log.Error("Inserted sensitive dex trie node", "owner", owner, "hash", hash)
}
}
// insertPreimage writes a new trie node pre-image to the memory database if it's
@ -349,7 +496,9 @@ func (db *Database) insertPreimage(hash common.Hash, preimage []byte) {
// node retrieves a cached trie node from memory, or returns nil if none can be
// found in the memory cache.
func (db *Database) node(hash common.Hash, cachegen uint16) node {
func (db *Database) node(owner common.Hash, hash common.Hash, cachegen uint16) node {
key := makeNodeKey(owner, hash)
// Retrieve the node from the clean cache if available
if db.cleans != nil {
if enc, err := db.cleans.Get(string(hash[:])); err == nil && enc != nil {
@ -360,14 +509,14 @@ func (db *Database) node(hash common.Hash, cachegen uint16) node {
}
// Retrieve the node from the dirty cache if available
db.lock.RLock()
dirty := db.dirties[hash]
dirty := db.dirties[key]
db.lock.RUnlock()
if dirty != nil {
return dirty.obj(hash, cachegen)
}
// Content unavailable in memory, attempt to retrieve from disk
enc, err := db.diskdb.Get(hash[:])
enc, err := db.diskdb.Get([]byte(key))
if err != nil || enc == nil {
return nil
}
@ -390,24 +539,34 @@ func (db *Database) Node(hash common.Hash) ([]byte, error) {
return enc, nil
}
}
// Retrieve the node from the dirty cache if available
db.lock.RLock()
dirty := db.dirties[hash]
db.lock.RUnlock()
// 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)
if dirty != nil {
return dirty.rlp(), nil
}
// Content unavailable in memory, attempt to retrieve from disk
enc, err := db.diskdb.Get(hash[:])
if err == nil && enc != nil {
if db.cleans != nil {
db.cleans.Set(string(hash[:]), enc)
memcacheCleanMissMeter.Mark(1)
memcacheCleanWriteMeter.Mark(int64(len(enc)))
db.lock.RLock()
dirty := db.dirties[key]
db.lock.RUnlock()
if dirty != nil {
return dirty.rlp(), nil
}
}
return enc, err
// Content unavailable in memory, attempt to retrieve from disk
enc, err := db.diskdb.Get([]byte(key))
if err == nil && enc != nil {
if db.cleans != nil {
db.cleans.Set(string(hash[:]), enc)
memcacheCleanMissMeter.Mark(1)
memcacheCleanWriteMeter.Mark(int64(len(enc)))
}
}
return enc, err
*/
return nil, nil
}
// preimage retrieves a cached trie node pre-image from memory. If it cannot be
@ -422,72 +581,83 @@ func (db *Database) preimage(hash common.Hash) ([]byte, error) {
return preimage, nil
}
// Content unavailable in memory, attempt to retrieve from disk
return db.diskdb.Get(db.secureKey(hash[:]))
return db.diskdb.Get(db.preimageKey(hash[:]))
}
// secureKey returns the database key for the preimage of key, as an ephemeral
// buffer. The caller must not hold onto the return value because it will become
// invalid on the next call.
func (db *Database) secureKey(key []byte) []byte {
buf := append(db.seckeybuf[:0], secureKeyPrefix...)
buf = append(buf, key...)
return buf
// preimageKey returns the database key for the preimage of key.
func (db *Database) preimageKey(key []byte) []byte {
return append(secureKeyPrefix, key...)
}
// Nodes retrieves the hashes of all the nodes cached within the memory database.
// This method is extremely expensive and should only be used to validate internal
// states in test code.
func (db *Database) Nodes() []common.Hash {
func (db *Database) Nodes() []string {
db.lock.RLock()
defer db.lock.RUnlock()
var hashes = make([]common.Hash, 0, len(db.dirties))
for hash := range db.dirties {
if hash != (common.Hash{}) { // Special case for "root" references/nodes
hashes = append(hashes, hash)
var keys = make([]string, 0, len(db.dirties))
for key := range db.dirties {
if key != metaRoot { // Special case for "root" references/nodes
keys = append(keys, key)
}
}
return hashes
return keys
}
// Reference adds a new reference from a parent node to a child node.
func (db *Database) Reference(child common.Hash, parent common.Hash) {
db.lock.RLock()
defer db.lock.RUnlock()
db.reference(child, parent)
}
// reference is the private locked version of Reference.
func (db *Database) reference(child common.Hash, parent common.Hash) {
// If the node does not exist, it's a node pulled from disk, skip
node, ok := db.dirties[child]
if !ok {
return
}
// If the reference already exists, only duplicate for roots
if db.dirties[parent].children == nil {
db.dirties[parent].children = make(map[common.Hash]uint16)
} else if _, ok = db.dirties[parent].children[child]; ok && parent != (common.Hash{}) {
return
}
node.parents++
db.dirties[parent].children[child]++
}
// Dereference removes an existing reference from a root node.
func (db *Database) Dereference(root common.Hash) {
// Sanity check to ensure that the meta-root is not removed
if root == (common.Hash{}) {
log.Error("Attempted to dereference the trie cache meta root")
return
// Reference adds a new reference from a parent node to a child node. We're going
// to break genericity here and assume that parent nodes are not owned (account
// trie) whereas child nodes may be owned (storage trie or bytecode).
func (db *Database) Reference(owner common.Hash, child common.Hash, parent common.Hash) {
// If pruning is enabled and running, something's very wrong
if db.pruner != nil && atomic.LoadUint32(&db.pruning) == 1 {
panic("pruner running during referencing")
}
db.lock.Lock()
defer db.lock.Unlock()
nodes, storage, start := len(db.dirties), db.dirtiesSize, time.Now()
db.dereference(root, common.Hash{})
// If the node does not exist, it's a node pulled from disk, skip
childKey := makeNodeKey(owner, child)
node, ok := db.dirties[childKey]
if !ok {
return
}
// If the reference already exists, only duplicate for roots
parentKey := makeNodeKey(common.Hash{}, parent)
if db.dirties[parentKey].children == nil {
db.dirties[parentKey].children = make(map[string]uint16)
} else if _, ok = db.dirties[parentKey].children[childKey]; ok && parent != (common.Hash{}) {
return
}
node.parents++
db.dirties[parentKey].children[childKey]++
}
// Dereference removes an existing reference from a root node.
func (db *Database) Dereference(root common.Hash) error {
// If pruning is enabled and running, something's very wrong
if db.pruner != nil && atomic.LoadUint32(&db.pruning) == 1 {
panic("pruner running during dereferencing")
}
// Sanity check to ensure that the meta-root is not removed
if root == (common.Hash{}) {
log.Error("Attempted to dereference the trie cache meta root")
return nil
}
// Obtain the write lock and garbage collect in-memory
db.lock.Lock()
defer db.lock.Unlock()
// Dereference the trie and accumulate prune targets if needed
nodes, storage, start := len(db.dirties), db.dirtiesSize, time.Now()
derefs := new([]*prunerTarget)
if err := db.dereference(common.Hash{}, root, common.Hash{}, common.Hash{}, nil, derefs); err != nil {
return err
}
if db.pruner != nil {
db.pruner.enqueue(*derefs)
}
db.gcnodes += uint64(nodes - len(db.dirties))
db.gcsize += storage - db.dirtiesSize
db.gctime += time.Since(start)
@ -496,59 +666,87 @@ func (db *Database) Dereference(root common.Hash) {
memcacheGCSizeMeter.Mark(int64(storage - db.dirtiesSize))
memcacheGCNodesMeter.Mark(int64(nodes - len(db.dirties)))
log.Debug("Dereferenced trie from memory database", "nodes", nodes-len(db.dirties), "size", storage-db.dirtiesSize, "time", time.Since(start),
"gcnodes", db.gcnodes, "gcsize", db.gcsize, "gctime", db.gctime, "livenodes", len(db.dirties), "livesize", db.dirtiesSize)
return nil
}
// dereference is the private locked version of Dereference.
func (db *Database) dereference(child common.Hash, parent common.Hash) {
func (db *Database) dereference(childOwner common.Hash, childHash common.Hash, parentOwner common.Hash, parentHash common.Hash, path []byte, derefs *[]*prunerTarget) error {
if childOwner == faultyOwner && childHash == faultyHash {
log.Error("Dereferencing sensitive dex trie node", "owner", childOwner, "hash", childHash)
}
// Dereference the parent-child
node := db.dirties[parent]
parentKey := makeNodeKey(parentOwner, parentHash)
parent := db.dirties[parentKey]
if node.children != nil && node.children[child] > 0 {
node.children[child]--
if node.children[child] == 0 {
delete(node.children, child)
childKey := makeNodeKey(childOwner, childHash)
if parent.children != nil && parent.children[childKey] > 0 {
parent.children[childKey]--
if parent.children[childKey] == 0 {
delete(parent.children, childKey)
}
}
// If the child does not exist, it's a previously committed node.
node, ok := db.dirties[child]
child, ok := db.dirties[childKey]
if !ok {
return
if db.pruner != nil {
*derefs = append(*derefs, &prunerTarget{
owner: childOwner,
hash: childHash,
path: common.CopyBytes(path),
})
}
return nil
}
if childOwner == faultyOwner && childHash == faultyHash {
log.Error("Dec. refcount on sensitive dex trie node", "rhash", crypto.Keccak256Hash(parent.rlp()).Hex(), "parent", hexutil.Encode(parent.rlp()), "owner", childOwner, "hash", childHash, "parents", fmt.Sprintf("%d->%d", child.parents, child.parents-1))
}
// If there are no more references to the child, delete it and cascade
if node.parents > 0 {
if child.parents > 0 {
// This is a special cornercase where a node loaded from disk (i.e. not in the
// memcache any more) gets reinjected as a new node (short node split into full,
// then reverted into short), causing a cached node to have no parents. That is
// no problem in itself, but don't make maxint parents out of it.
node.parents--
child.parents--
}
if node.parents == 0 {
if child.parents == 0 {
// Remove the node from the flush-list
switch child {
switch childKey {
case db.oldest:
db.oldest = node.flushNext
db.dirties[node.flushNext].flushPrev = common.Hash{}
db.oldest = child.flushNext
db.dirties[child.flushNext].flushPrev = metaRoot
case db.newest:
db.newest = node.flushPrev
db.dirties[node.flushPrev].flushNext = common.Hash{}
db.newest = child.flushPrev
db.dirties[child.flushPrev].flushNext = metaRoot
default:
db.dirties[node.flushPrev].flushNext = node.flushNext
db.dirties[node.flushNext].flushPrev = node.flushPrev
db.dirties[child.flushPrev].flushNext = child.flushNext
db.dirties[child.flushNext].flushPrev = child.flushPrev
}
// Dereference all children and delete the node
for _, hash := range node.childs() {
db.dereference(hash, child)
child.iterateRefs(path, func(path []byte, hash common.Hash) error {
db.dereference(childOwner, hash, childOwner, childHash, path, derefs)
return nil
})
for key := range child.children {
owner, hash := splitNodeKey(key)
db.dereference(owner, hash, childOwner, childHash, nil, derefs)
}
delete(db.dirties, childKey)
db.dirtiesSize -= common.StorageSize(common.HashLength + int(child.size))
if childOwner == faultyOwner && childHash == faultyHash {
log.Error("Dereferenced sensitive dex trie node", "owner", childOwner, "hash", childHash)
}
delete(db.dirties, child)
db.dirtiesSize -= common.StorageSize(common.HashLength + int(node.size))
}
return nil
}
// Cap iteratively flushes old but still referenced trie nodes until the total
// memory usage goes below the given threshold.
func (db *Database) Cap(limit common.StorageSize) error {
// If pruning is enabled and running, something's very wrong
if db.pruner != nil && atomic.LoadUint32(&db.pruning) == 1 {
panic("pruner running during capping")
}
// Create a database batch to flush persistent data out. It is important that
// outside code doesn't see an inconsistent state (referenced data removed from
// memory cache during commit but not yet in persistent storage). This is ensured
@ -568,7 +766,7 @@ func (db *Database) Cap(limit common.StorageSize) error {
flushPreimages := db.preimagesSize > 4*1024*1024
if flushPreimages {
for hash, preimage := range db.preimages {
if err := batch.Put(db.secureKey(hash[:]), preimage); err != nil {
if err := batch.Put(db.preimageKey(hash[:]), preimage); err != nil {
log.Error("Failed to commit preimage from trie database", "err", err)
db.lock.RUnlock()
return err
@ -584,13 +782,16 @@ func (db *Database) Cap(limit common.StorageSize) error {
}
// Keep committing nodes from the flush-list until we're below allowance
oldest := db.oldest
for size > limit && oldest != (common.Hash{}) {
for size > limit && oldest != metaRoot {
// Fetch the oldest referenced node and push into the batch
node := db.dirties[oldest]
if err := batch.Put(oldest[:], node.rlp()); err != nil {
if err := batch.Put([]byte(oldest), node.rlp()); err != nil {
db.lock.RUnlock()
return err
}
if oldest == makeNodeKey(faultyOwner, faultyHash) {
log.Error("Flush sensitive dex trie node", "owner", faultyOwner, "hash", faultyHash)
}
// If we exceeded the ideal batch size, commit and reset
if batch.ValueSize() >= ethdb.IdealBatchSize {
if err := batch.Write(); err != nil {
@ -630,8 +831,8 @@ func (db *Database) Cap(limit common.StorageSize) error {
db.dirtiesSize -= common.StorageSize(common.HashLength + int(node.size))
}
if db.oldest != (common.Hash{}) {
db.dirties[db.oldest].flushPrev = common.Hash{}
if db.oldest != metaRoot {
db.dirties[db.oldest].flushPrev = metaRoot
}
db.flushnodes += uint64(nodes - len(db.dirties))
db.flushsize += storage - db.dirtiesSize
@ -641,8 +842,8 @@ func (db *Database) Cap(limit common.StorageSize) error {
memcacheFlushSizeMeter.Mark(int64(storage - db.dirtiesSize))
memcacheFlushNodesMeter.Mark(int64(nodes - len(db.dirties)))
log.Debug("Persisted nodes from memory database", "nodes", nodes-len(db.dirties), "size", storage-db.dirtiesSize, "time", time.Since(start),
"flushnodes", db.flushnodes, "flushsize", db.flushsize, "flushtime", db.flushtime, "livenodes", len(db.dirties), "livesize", db.dirtiesSize)
log.Debug("Persisted nodes from memory database", "nodes", nodes-len(db.dirties), "size", storage-db.dirtiesSize, "time", common.PrettyDuration(time.Since(start)),
"flnodes", db.flushnodes, "flsize", db.flushsize, "fltime", common.PrettyDuration(db.flushtime), "livenodes", len(db.dirties), "livesize", db.dirtiesSize)
return nil
}
@ -652,6 +853,10 @@ func (db *Database) Cap(limit common.StorageSize) error {
//
// As a side effect, all pre-images accumulated up to this point are also written.
func (db *Database) Commit(node common.Hash, report bool) error {
// If pruning is enabled and running, something's very wrong
if db.pruner != nil && atomic.LoadUint32(&db.pruning) == 1 {
panic("pruner running during committing")
}
// Create a database batch to flush persistent data out. It is important that
// outside code doesn't see an inconsistent state (referenced data removed from
// memory cache during commit but not yet in persistent storage). This is ensured
@ -663,7 +868,7 @@ func (db *Database) Commit(node common.Hash, report bool) error {
// Move all of the accumulated preimages into a write batch
for hash, preimage := range db.preimages {
if err := batch.Put(db.secureKey(hash[:]), preimage); err != nil {
if err := batch.Put(db.preimageKey(hash[:]), preimage); err != nil {
log.Error("Failed to commit preimage from trie database", "err", err)
db.lock.RUnlock()
return err
@ -678,7 +883,7 @@ func (db *Database) Commit(node common.Hash, report bool) error {
}
// Move the trie itself into the batch, flushing if enough data is accumulated
nodes, storage := len(db.dirties), db.dirtiesSize
if err := db.commit(node, batch); err != nil {
if err := db.commit(common.Hash{}, node, batch); err != nil {
log.Error("Failed to commit trie from trie database", "err", err)
db.lock.RUnlock()
return err
@ -698,7 +903,7 @@ func (db *Database) Commit(node common.Hash, report bool) error {
db.preimages = make(map[common.Hash][]byte)
db.preimagesSize = 0
db.uncache(node)
db.uncache(common.Hash{}, node)
memcacheCommitTimeTimer.Update(time.Since(start))
memcacheCommitSizeMeter.Mark(int64(storage - db.dirtiesSize))
@ -708,29 +913,39 @@ func (db *Database) Commit(node common.Hash, report bool) error {
if !report {
logger = log.Debug
}
logger("Persisted trie from memory database", "nodes", nodes-len(db.dirties)+int(db.flushnodes), "size", storage-db.dirtiesSize+db.flushsize, "time", time.Since(start)+db.flushtime,
"gcnodes", db.gcnodes, "gcsize", db.gcsize, "gctime", db.gctime, "livenodes", len(db.dirties), "livesize", db.dirtiesSize)
logger("Persisted trie from memory database", "nodes", nodes-len(db.dirties)+int(db.flushnodes), "size", storage-db.dirtiesSize+db.flushsize, "time", common.PrettyDuration(time.Since(start)+db.flushtime),
"gcnodes", db.gcnodes, "gcsize", db.gcsize, "gctime", common.PrettyDuration(db.gctime), "prnodes", db.prunenodes, "prsize", db.prunesize, "prtime", common.PrettyDuration(db.prunetime),
"linodes", len(db.dirties), "lisize", db.dirtiesSize)
// Reset the garbage collection statistics
db.gcnodes, db.gcsize, db.gctime = 0, 0, 0
db.prunenodes, db.prunesize, db.prunetime = 0, 0, 0
db.flushnodes, db.flushsize, db.flushtime = 0, 0, 0
return nil
}
// commit is the private locked version of Commit.
func (db *Database) commit(hash common.Hash, batch ethdb.Batch) error {
func (db *Database) commit(owner common.Hash, hash common.Hash, batch ethdb.Batch) error {
// If the node does not exist, it's a previously committed node
node, ok := db.dirties[hash]
key := makeNodeKey(owner, hash)
node, ok := db.dirties[key]
if !ok {
return nil
}
for _, child := range node.childs() {
if err := db.commit(child, batch); err != nil {
if err := node.iterateRefs(nil, func(path []byte, child common.Hash) error {
return db.commit(owner, child, batch)
}); err != nil {
return err
}
for child := range node.children {
owner, hash := splitNodeKey(child)
if err := db.commit(owner, hash, batch); err != nil {
return err
}
}
if err := batch.Put(hash[:], node.rlp()); err != nil {
if err := batch.Put([]byte(key), node.rlp()); err != nil {
return err
}
// If we've reached an optimal batch size, commit and start over
@ -747,29 +962,35 @@ func (db *Database) commit(hash common.Hash, batch ethdb.Batch) error {
// persisted trie is removed from the cache. The reason behind the two-phase
// commit is to ensure consistent data availability while moving from memory
// to disk.
func (db *Database) uncache(hash common.Hash) {
func (db *Database) uncache(owner common.Hash, hash common.Hash) {
// If the node does not exist, we're done on this path
node, ok := db.dirties[hash]
key := makeNodeKey(owner, hash)
node, ok := db.dirties[key]
if !ok {
return
}
// Node still exists, remove it from the flush-list
switch hash {
switch key {
case db.oldest:
db.oldest = node.flushNext
db.dirties[node.flushNext].flushPrev = common.Hash{}
db.dirties[node.flushNext].flushPrev = metaRoot
case db.newest:
db.newest = node.flushPrev
db.dirties[node.flushPrev].flushNext = common.Hash{}
db.dirties[node.flushPrev].flushNext = metaRoot
default:
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
for _, child := range node.childs() {
db.uncache(child)
node.iterateRefs(nil, func(path []byte, child common.Hash) error {
db.uncache(owner, child)
return nil
})
for child := range node.children {
db.uncache(splitNodeKey(child))
}
delete(db.dirties, hash)
delete(db.dirties, key)
db.dirtiesSize -= common.StorageSize(common.HashLength + int(node.size))
}
@ -794,17 +1015,18 @@ func (db *Database) Size() (common.StorageSize, common.StorageSize) {
// This method is extremely CPU and memory intensive, only use when must.
func (db *Database) verifyIntegrity() {
// Iterate over all the cached nodes and accumulate them into a set
reachable := map[common.Hash]struct{}{{}: {}}
reachable := map[string]struct{}{metaRoot: struct{}{}}
for child := range db.dirties[common.Hash{}].children {
db.accumulate(child, reachable)
for key := range db.dirties[metaRoot].children {
_, root := splitNodeKey(key)
db.accumulate(common.Hash{}, root, reachable)
}
// Find any unreachable but cached nodes
var unreachable []string
for hash, node := range db.dirties {
if _, ok := reachable[hash]; !ok {
for key, node := range db.dirties {
if _, ok := reachable[key]; !ok {
unreachable = append(unreachable, fmt.Sprintf("%x: {Node: %v, Parents: %d, Prev: %x, Next: %x}",
hash, node.node, node.parents, node.flushPrev, node.flushNext))
key, node.node, node.parents, node.flushPrev, node.flushNext))
}
}
if len(unreachable) != 0 {
@ -812,18 +1034,25 @@ func (db *Database) verifyIntegrity() {
}
}
// accumulate iterates over the trie defined by hash and accumulates all the
// cached children found in memory.
func (db *Database) accumulate(hash common.Hash, reachable map[common.Hash]struct{}) {
// accumulate iterates over the trie defined by owner:hash and accumulates all
// the cached children found in memory.
func (db *Database) accumulate(owner common.Hash, hash common.Hash, reachable map[string]struct{}) {
// Mark the node reachable if present in the memory cache
node, ok := db.dirties[hash]
key := makeNodeKey(owner, hash)
node, ok := db.dirties[key]
if !ok {
return
}
reachable[hash] = struct{}{}
reachable[key] = struct{}{}
// Iterate over all the children and accumulate them too
for _, child := range node.childs() {
db.accumulate(child, reachable)
node.iterateRefs(nil, func(path []byte, hash common.Hash) error {
db.accumulate(owner, hash, reachable)
return nil
})
for key := range node.children {
owner, hash := splitNodeKey(key)
db.accumulate(owner, hash, reachable)
}
}

25
trie/hasher.go
View file

@ -31,6 +31,7 @@ type hasher struct {
cachegen uint16
cachelimit uint16
onleaf LeafCallback
owner common.Hash
}
// keccakState wraps sha3.state. In addition to the usual hash methods, it also supports
@ -62,9 +63,9 @@ var hasherPool = sync.Pool{
},
}
func newHasher(cachegen, cachelimit uint16, onleaf LeafCallback) *hasher {
func newHasher(owner common.Hash, cachegen, cachelimit uint16, onleaf LeafCallback) *hasher {
h := hasherPool.Get().(*hasher)
h.cachegen, h.cachelimit, h.onleaf = cachegen, cachelimit, onleaf
h.owner, h.cachegen, h.cachelimit, h.onleaf = owner, cachegen, cachelimit, onleaf
return h
}
@ -74,7 +75,7 @@ func returnHasherToPool(h *hasher) {
// hash collapses a node down into a hash node, also returning a copy of the
// original node initialized with the computed hash to replace the original one.
func (h *hasher) hash(n node, db *Database, force bool) (node, node, error) {
func (h *hasher) hash(path []byte, n node, db *Database, force bool) (node, node, error) {
// If we're not storing the node, just hashing, use available cached data
if hash, dirty := n.cache(); hash != nil {
if db == nil {
@ -91,11 +92,11 @@ func (h *hasher) hash(n node, db *Database, force bool) (node, node, error) {
}
}
// Trie not processed yet or needs storage, walk the children
collapsed, cached, err := h.hashChildren(n, db)
collapsed, cached, err := h.hashChildren(path, n, db)
if err != nil {
return hashNode{}, n, err
}
hashed, err := h.store(collapsed, db, force)
hashed, err := h.store(path, collapsed, db, force)
if err != nil {
return hashNode{}, n, err
}
@ -121,7 +122,7 @@ func (h *hasher) hash(n node, db *Database, force bool) (node, node, error) {
// hashChildren replaces the children of a node with their hashes if the encoded
// size of the child is larger than a hash, returning the collapsed node as well
// as a replacement for the original node with the child hashes cached in.
func (h *hasher) hashChildren(original node, db *Database) (node, node, error) {
func (h *hasher) hashChildren(path []byte, original node, db *Database) (node, node, error) {
var err error
switch n := original.(type) {
@ -132,7 +133,7 @@ func (h *hasher) hashChildren(original node, db *Database) (node, node, error) {
cached.Key = common.CopyBytes(n.Key)
if _, ok := n.Val.(valueNode); !ok {
collapsed.Val, cached.Val, err = h.hash(n.Val, db, false)
collapsed.Val, cached.Val, err = h.hash(append(path, n.Key...), n.Val, db, false)
if err != nil {
return original, original, err
}
@ -145,7 +146,7 @@ func (h *hasher) hashChildren(original node, db *Database) (node, node, error) {
for i := 0; i < 16; i++ {
if n.Children[i] != nil {
collapsed.Children[i], cached.Children[i], err = h.hash(n.Children[i], db, false)
collapsed.Children[i], cached.Children[i], err = h.hash(append(path, byte(i)), n.Children[i], db, false)
if err != nil {
return original, original, err
}
@ -163,7 +164,7 @@ func (h *hasher) hashChildren(original node, db *Database) (node, node, error) {
// store hashes the node n and if we have a storage layer specified, it writes
// the key/value pair to it and tracks any node->child references as well as any
// node->external trie references.
func (h *hasher) store(n node, db *Database, force bool) (node, error) {
func (h *hasher) store(path []byte, n node, db *Database, force bool) (node, error) {
// Don't store hashes or empty nodes.
if _, isHash := n.(hashNode); n == nil || isHash {
return n, nil
@ -187,7 +188,7 @@ func (h *hasher) store(n node, db *Database, force bool) (node, error) {
hash := common.BytesToHash(hash)
db.lock.Lock()
db.insert(hash, h.tmp, n)
db.insert(h.owner, hash, h.tmp, n)
db.lock.Unlock()
// Track external references from account->storage trie
@ -195,12 +196,12 @@ func (h *hasher) store(n node, db *Database, force bool) (node, error) {
switch n := n.(type) {
case *shortNode:
if child, ok := n.Val.(valueNode); ok {
h.onleaf(child, hash)
h.onleaf(common.BytesToHash(hexToKeybytes(append(path, compactToHex(n.Key)...))), child, hash)
}
case *fullNode:
for i := 0; i < 16; i++ {
if child, ok := n.Children[i].(valueNode); ok {
h.onleaf(child, hash)
h.onleaf(common.BytesToHash(hexToKeybytes(append(path, byte(i)))), child, hash)
}
}
}

7
trie/iterator.go
View file

@ -180,15 +180,14 @@ func (it *nodeIterator) LeafBlob() []byte {
func (it *nodeIterator) LeafProof() [][]byte {
if len(it.stack) > 0 {
if _, ok := it.stack[len(it.stack)-1].node.(valueNode); ok {
hasher := newHasher(0, 0, nil)
hasher := newHasher(common.Hash{}, 0, 0, nil)
defer returnHasherToPool(hasher)
proofs := make([][]byte, 0, len(it.stack))
for i, item := range it.stack[:len(it.stack)-1] {
// Gather nodes that end up as hash nodes (or the root)
node, _, _ := hasher.hashChildren(item.node, nil)
hashed, _ := hasher.store(node, nil, false)
node, _, _ := hasher.hashChildren(nil, item.node, nil)
hashed, _ := hasher.store(nil, node, nil, false)
if _, ok := hashed.(hashNode); ok || i == 0 {
enc, _ := rlp.EncodeToBytes(node)
proofs = append(proofs, enc)

2
trie/node.go
View file

@ -120,7 +120,7 @@ func (n valueNode) fstring(ind string) string {
func mustDecodeNode(hash, buf []byte, cachegen uint16) node {
n, err := decodeNode(hash, buf, cachegen)
if err != nil {
panic(fmt.Sprintf("node %x: %v", hash, err))
panic(fmt.Sprintf("node %x (%x): %v", hash, buf, err))
}
return n
}

6
trie/proof.go
View file

@ -65,14 +65,14 @@ func (t *Trie) Prove(key []byte, fromLevel uint, proofDb ethdb.Writer) error {
panic(fmt.Sprintf("%T: invalid node: %v", tn, tn))
}
}
hasher := newHasher(0, 0, nil)
hasher := newHasher(common.Hash{}, 0, 0, nil)
defer returnHasherToPool(hasher)
for i, n := range nodes {
// Don't bother checking for errors here since hasher panics
// if encoding doesn't work and we're not writing to any database.
n, _, _ = hasher.hashChildren(n, nil)
hn, _ := hasher.store(n, nil, false)
n, _, _ = hasher.hashChildren(nil, n, nil)
hn, _ := hasher.store(nil, n, nil, false)
if hash, ok := hn.(hashNode); ok || i == 0 {
// If the node's database encoding is a hash (or is the
// root node), it becomes a proof element.

518
trie/pruning.go Normal file
View file

@ -0,0 +1,518 @@
// Copyright 2019 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 (
"bytes"
"fmt"
"math/big"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/rlp"
"github.com/karalabe/cookiejar/collections/deque"
)
var (
memcachePruneTimeTimer = metrics.NewRegisteredResettingTimer("trie/memcache/prune/time", nil)
memcachePruneNodesMeter = metrics.NewRegisteredMeter("trie/memcache/prune/nodes", nil)
memcachePruneSizeMeter = metrics.NewRegisteredMeter("trie/memcache/prune/size", nil)
memcachePruneAssignHistogram = metrics.NewRegisteredHistogram("trie/memcache/prune/assign", nil, metrics.NewExpDecaySample(1028, 0.015))
memcachePruneAssignDupHistogram = metrics.NewRegisteredHistogram("trie/memcache/prune/assigndup", nil, metrics.NewExpDecaySample(1028, 0.015))
memcachePruneRemainHistogram = metrics.NewRegisteredHistogram("trie/memcache/prune/remain", nil, metrics.NewExpDecaySample(1028, 0.015))
memcachePruneRemainDupHistogram = metrics.NewRegisteredHistogram("trie/memcache/prune/remaindup", nil, metrics.NewExpDecaySample(1028, 0.015))
memcachePruneQueueHistogram = metrics.NewRegisteredHistogram("trie/memcache/prune/queue", nil, metrics.NewExpDecaySample(1028, 0.015))
)
// pruner is responsible for pruning the state trie based on liveness checks
// whenever the in-memory garbage collector attempt to dereference a node from
// disk.
//
// Note, the pruner is not a standalone construct, rather an extension to the
// trie database. No attempt was made to separate the API surface and make one
// a disjoint client of the other.
type pruner struct {
db *Database // Trie database for accessing dirty and clean data
taskCh chan []*prunerTarget // Task queue receiving the pruning targets to delete
abortCh chan chan struct{} // Notification channel to terminate the pruner
resumeCh chan chan struct{} // Notification channel to resume the pruner
interrupt uint32 // Signals to a running deep pruning to suspend itself
}
// prunerTarget represents a single marked target for potential pruning.
type prunerTarget struct {
owner common.Hash // Owner account hash of the node to delete
path []byte // Patricia path leading to this node
hash common.Hash // Hash of the node to delete
}
// newPruner creates a new background trie pruner to delete unreferenced nodes
// whenever the tries are not being actively written.
func newPruner(db *Database) *pruner {
p := &pruner{
db: db,
taskCh: make(chan []*prunerTarget),
abortCh: make(chan chan struct{}),
resumeCh: make(chan chan struct{}),
}
go p.loop()
return p
}
// enqueue adds a batch of potential prune targets to the removal queue to be
// inspected and removed from the database if deemed unreferenced by recentMeter.
// and snapshot tries.
//
// It's important to queue in batches as a single block might enque hundreds or
// thousands of targets. Queueing individually entails a huge performance hit.
func (p *pruner) enqueue(targets []*prunerTarget) {
p.taskCh <- targets
}
// resume (re)starts the pruning, locking the dirty caches for reads to prevent
// trie nodes going missing due to concurrent pruning/referencing.
//
// Note, calling resume on an already running pruner will deadlock!
func (p *pruner) resume() {
// The prumer might have been interrupted previously, so we need to ensure the
// interrut is cleared before requesting a resumption. This could be done by the
// pruner ron loop too, but figured it might be cleaner to set the interrupt at
// the same scope as with `pause`,
atomic.StoreUint32(&p.interrupt, 0)
// We *must* wait for the pruner to obtain the lock, otherwise the caller might
// race forward and lock the database for writing, messing up the state machine.
ch := make(chan struct{})
p.resumeCh <- ch
<-ch
}
// pause signals the pruner to interrupt its operation and release its held lock.
// This is needed for the block processor to obtain a write lock on the dirty
// caches, which are otherwise held hostage by the pruner.
//
// Note, calling pause on a non-running pruner will panic!
func (p *pruner) pause() {
// Notify the pruner to abort right now.
atomic.StoreUint32(&p.interrupt, 1)
}
// terminate signals the pruner to finish all remaining tasks and permanently
// release all locks and clean itself up.
//
// Note, calling terminate on a non-running pruner will panic!
func (p *pruner) terminate() {
// Signal to the pruner that it should terminate itself gracefully and wait for
// it to confirm before pulling the rug from underneath.
ch := make(chan struct{})
p.abortCh <- ch
<-ch
}
// loop is the pruner background gorutineo that waits for pruning targets the be
// added, causing liveness checks and potentially database deletions in response.
func (p *pruner) loop() {
var (
tasks = deque.New() // Queue of trie nodes queued for potential pruning
taskset = make(map[string]struct{}) // Set of trie nodes queued to prevent duplication
tries []*traverser // Individual trie traversers for liveness checks
quit chan struct{} // Quit signal channel when termination is requested
batch = p.db.diskdb.NewBatch() // Create a write batch to minimize thrashing
)
// Wait for different events and process them accordingly
for {
select {
case targets := <-p.taskCh:
// New task received, queue it up. We will not start immediately processing
// this as the enqueueing is done whilst doing in-memory garbage collection,
// so the dirty caches are locked for writing.
duplicates := 0
for _, task := range targets {
key := makeNodeKey(task.owner, task.hash)
if _, exists := taskset[key]; exists {
duplicates++
continue
}
tasks.PushRight(task)
taskset[key] = struct{}{}
}
memcachePruneQueueHistogram.Update(int64(tasks.Size()))
memcachePruneAssignHistogram.Update(int64(len(targets)))
memcachePruneAssignDupHistogram.Update(int64(duplicates))
case ch := <-p.resumeCh:
// Pruner was requested to resume operation. Obtain the necessary locks to
// prevent the block processor for modifying the dirty caches, but allow any
// goroutines to still read the data.
if tasks.Size() == 0 {
ch <- struct{}{} // signal back, but nothing to do really
continue
}
p.db.lock.RLock()
ch <- struct{}{} // signal back that the lock was obtained
// Ensure the traversers are pointing to the currently live tries. Usually
// after each pause/resume cycle, one (new block) or two (new snapshot) tries
// get swapped out.
tries = nil // cheat a bit for now and just reconstruct them
for key := range p.db.dirties[metaRoot].children {
_, root := splitNodeKey(key)
tries = append(tries, &traverser{
db: p.db,
state: &traverserState{hash: root, node: hashNode(root[:])},
})
}
for hash := range p.db.noprune {
tries = append(tries, &traverser{
db: p.db,
state: &traverserState{hash: hash, node: hashNode(common.CopyBytes(hash[:]))}, // need closure!
})
}
// Process the tasks until an interrupt arrives
start, nodes, size := time.Now(), p.db.prunenodes, p.db.prunesize
for !tasks.Empty() {
// Delete this particular task from the deduplication set
task := tasks.PopLeft().(*prunerTarget)
delete(taskset, makeNodeKey(task.owner, task.hash))
// Prune the target and reschedule any interrupted sub-tasks
remain := p.prune(task.owner, task.hash, task.path, taskset, tries, batch)
if atomic.LoadUint32(&p.interrupt) == 1 {
duplicates := 0
for j := len(remain) - 1; j >= 0; j-- { // reverse to keep the depth priority
// Dedup already scheduled tasks, no need to prune twice
key := makeNodeKey(remain[j].owner, remain[j].hash)
if _, exist := taskset[key]; exist {
duplicates++
continue
}
// Reschedule (high priority) anything that's not a duplicate
tasks.PushLeft(remain[j])
taskset[key] = struct{}{}
}
memcachePruneQueueHistogram.Update(int64(tasks.Size()))
memcachePruneRemainHistogram.Update(int64(len(remain)))
memcachePruneRemainDupHistogram.Update(int64(duplicates))
break
}
}
// If all tasks have been procesed, get rid of any allocated task slice and
// terminate the runner pathway.
if tasks.Empty() {
tasks.Reset()
memcachePruneQueueHistogram.Update(0)
memcachePruneRemainHistogram.Update(0)
memcachePruneRemainDupHistogram.Update(0)
}
// Update all the stats with the results until now
memcachePruneNodesMeter.Mark(int64(p.db.prunenodes - nodes))
memcachePruneSizeMeter.Mark(int64(p.db.prunesize - size))
memcachePruneTimeTimer.Update(time.Since(start))
p.db.prunetime += time.Since(start)
// Push any change to disk
if err := batch.Write(); err != nil {
log.Crit("Failed to flush pruned nodes", "err", err)
}
batch.Reset()
// Relinquish the lock to any writer. We can't do this earlier to avoid data
// races between this goroutine deleting the same node that some other one is
// attempting to put back.
p.db.lock.RUnlock()
// If we're actually shutting down, clean up everything
if quit != nil {
quit <- struct{}{}
return
}
case quit = <-p.abortCh:
// Pruner was requetsed to terminate. Since termination doesn't interrupt, we
// can at this point safely assume everything was pruned.
quit <- struct{}{}
return
}
}
}
var faultyOwner = common.HexToHash("0x9f13f88230a70de90ed5fa41ba35a5fb78bc55d11cc9406f17d314fb67047ac7")
var faultyHash = common.HexToHash("0x5610d8d5e4056edad0db0743616df01c0911675c5fc5604c8967312a4961cf72")
// prune deletes a trie node from disk if there are no more live references to
// it, cascading until all dangling nodes are removed. If the pruner's interrupt
// has been triggered (block processing pending), the remaining nodes are bubbled
// up to the caller to reschedule later.
func (p *pruner) prune(owner common.Hash, hash common.Hash, path []byte, taskset map[string]struct{}, tries []*traverser, batch ethdb.Batch) []*prunerTarget {
if owner == faultyOwner && hash == faultyHash {
log.Error("Evaluating sensitive dex trie node", "owner", owner, "hash", hash, "path", hexutil.Encode(path))
}
// If the node is already queued for pruning, don't duplicate any effort on it
key := makeNodeKey(owner, hash)
if _, ok := taskset[key]; ok {
return nil
}
// If the node is still live in the memory cache, it's still referenced so we
// can abort. This case is important when and old trie being pruned references
// a new node (maybe that node was recreated since), since currently live nodes
// are stored expanded, not as hashes.
if p.db.dirties[key] != nil {
return nil
}
// Iterate over all the live tries and check node liveliness
crosspath := path
if owner != (common.Hash{}) {
crosspath = append(append(keybytesToHex(owner[:]), 0xff), crosspath...)
}
unrefs := make(map[common.Hash]bool)
for _, trie := range tries {
// If the node is still live, abort
if trie.live(owner, hash, crosspath, unrefs) {
return nil
}
// Node dead in this trie, cache the result for subsequent traversals
trie.unref(2, unrefs)
}
// Dead node found, delete it from the database
dead := []byte(key)
blob, err := p.db.diskdb.Get(dead)
if blob == nil || err != nil {
// Node already deleted by something else, happens with delayed pruning
//log.Error("Missing prune target", "owner", owner, "hash", hash, "path", fmt.Sprintf("%x", path))
return nil
}
node := mustDecodeNode(hash[:], blob, 0)
// Prune the node and its children if it's not a bytecode blob
if owner == faultyOwner && hash == faultyHash {
log.Error("Deleting sensitive dex trie node", "owner", owner, "hash", hash, "path", hexutil.Encode(path))
}
p.db.cleans.Delete(string(hash[:]))
batch.Delete(dead)
p.db.prunenodes++
p.db.prunesize += common.StorageSize(len(blob))
var remain []*prunerTarget
iterateRefs(node, path, func(path []byte, hash common.Hash) error {
// If the pruner was interrupted, accumulate the remaining targets
if atomic.LoadUint32(&p.interrupt) == 1 {
remain = append(remain, &prunerTarget{owner: owner, hash: hash, path: common.CopyBytes(path)})
return nil
}
// Pruning not interrupted until now, attempt to process children too. It's
// fine to assign the result directly to the `remain` slice because it's nil
// anyway until the interrupt triggers.
remain = p.prune(owner, hash, path, taskset, tries, batch)
return nil
})
return remain
}
// traverser is a stateful trie traversal data structure used by the pruner to
// verify the liveness of a node within a specific trie. The reason for having
// a separate data structure is to allow reusing previous traversals to check
// the liveness of nested nodes (i.e. entire subtried during pruning).
type traverser struct {
db *Database // Trie database for accessing dirty and clean data
state *traverserState // Leftover state from the previous traversals
}
// traverserState is the internal state of a trie traverser.
type traverserState struct {
parent *traverserState // Parent traverser to allow backtracking
prefix []byte // Path leading up to the root of this traverser
node node // Trie node where this traverser is currently at
hash common.Hash // Hash of the trie node at the traversed position
}
// live checks whether the trie iterated by this traverser contains the hashnode
// at the given path, minimizing data access and processing by reusing previous
// state instead of starting fresh.
//
// The path is a full canonical path from the account trie root down to the node
// potentially crossing over into a storage trie. The account and storage trie
// paths are separated by a 0xff byte (nibbles range from 0x00-0x10). This byte
// is needed to differentiate between the leaf of the account trie and the root
// of a storage trie (which otherwise would have the same traversal path).
func (t *traverser) live(owner common.Hash, hash common.Hash, path []byte, unrefs map[common.Hash]bool) bool {
// Rewind the traverser until it's prefix is actually a prefix of the path
for !bytes.HasPrefix(path, t.state.prefix) {
t.state = t.state.parent
}
// Short circuit the liveness check if we already covered this prefix (if this
// prefix path was not yet seen in previous tries, no parent could have been
// seen either, so no point in checkin upwards further than the first hash).
for state := t.state; state != nil; state = state.parent {
if state.hash != (common.Hash{}) {
if unrefs[state.hash] {
return false
}
break
}
}
// Traverse downward until the prefix matches the path completely
path = path[len(t.state.prefix):]
for len(path) > 0 {
// If we're at a hash node, expand before continuing
if n, ok := t.state.node.(hashNode); ok {
// Short circuit if we already encountered this node
t.state.hash = common.BytesToHash(n)
if unrefs[t.state.hash] {
return false
}
// Generate the database key for this hash node
var key string
if len(t.state.prefix) < 2*common.HashLength {
key = makeNodeKey(common.Hash{}, t.state.hash)
} else {
key = makeNodeKey(owner, t.state.hash)
}
// Replace the node in the traverser with the expanded one
if enc, err := t.db.cleans.Get(string(t.state.hash[:])); err == nil && enc != nil {
t.state.node = mustDecodeNode(t.state.hash[:], enc, 0)
} else if node := t.db.dirties[key]; node != nil {
t.state.node = node.node
} else {
blob, err := t.db.diskdb.Get([]byte(key))
if blob == nil || err != nil {
log.Error("Missing referenced node", "owner", owner, "hash", t.state.hash.Hex(), "path", fmt.Sprintf("%x%x", t.state.prefix, path))
return false
//panic(fmt.Sprintf("missing referenced node %x (searching for %x:%x at %x%x)", key, owner, t.state.hash, t.state.prefix, path))
}
t.state.node = mustDecodeNode(t.state.hash[:], blob, 0)
t.db.cleans.Set(string(t.state.hash[:]), blob)
}
}
// If we reached an account node, extract the storage trie root to continue on
if path[0] == 0xff {
// Retrieve the storage trie root and abort if empty
if have, ok := t.state.node.(valueNode); ok {
var account struct {
Nonce uint64
Balance *big.Int
Root common.Hash
CodeHash []byte
}
if err := rlp.DecodeBytes(have, &account); err != nil {
panic(err)
}
if account.Root == emptyRoot {
return false
}
// Create a new nesting in the traversal and continue on that depth
t.state, path = &traverserState{
parent: t.state,
prefix: append(t.state.prefix, 0xff),
node: hashNode(account.Root[:]),
}, path[1:]
continue
}
panic(fmt.Sprintf("liveness check path swap terminated on non value node: %T", t.state.node))
}
// Descend into the trie following the specified path. This code segment must
// be able to handle both simplified raw nodes kept in this cache as well as
// cold nodes loaded directly from disk.
switch n := t.state.node.(type) {
case *rawShortNode:
if prefixLen(n.Key, path) == len(n.Key) {
t.state, path = &traverserState{
parent: t.state,
prefix: append(t.state.prefix, path[:len(n.Key)]...),
node: n.Val,
}, path[len(n.Key):]
continue
}
return false
case *shortNode:
if prefixLen(n.Key, path) == len(n.Key) {
t.state, path = &traverserState{
parent: t.state,
prefix: append(t.state.prefix, path[:len(n.Key)]...),
node: n.Val,
}, path[len(n.Key):]
continue
}
return false
case rawFullNode:
if child := n[path[0]]; child != nil {
t.state, path = &traverserState{
parent: t.state,
prefix: append(t.state.prefix, path[0]),
node: child,
}, path[1:]
continue
}
return false
case *fullNode:
if child := n.Children[path[0]]; child != nil {
t.state, path = &traverserState{
parent: t.state,
prefix: append(t.state.prefix, path[0]),
node: child,
}, path[1:]
continue
}
return false
default:
panic(fmt.Sprintf("unknown node type: %T", n))
}
}
// The prefix should match perfectly here, check if the hashes matches
if t.state.hash != (common.Hash{}) { // expanded/cached hash node
return t.state.hash == hash
}
if have, ok := t.state.node.(hashNode); ok { // collapsed hash node
t.state.hash = common.BytesToHash(have)
return t.state.hash == hash
}
return false
}
// unref marks the current traversal nodes as *not* containing the specific trie
// node having been searched for. It is used by searches in subsequent tries to
// avoid reiterating the exact same sub-tries.
func (t *traverser) unref(count int, unrefs map[common.Hash]bool) {
state := t.state
for state != nil && count > 0 {
// If we've found a hash node, store it as a subresult
if state.hash != (common.Hash{}) {
unrefs[state.hash] = true
count--
}
// Traverse further up to the next hash node
state = state.parent
}
}

20
trie/secure_trie.go
View file

@ -52,10 +52,26 @@ type SecureTrie struct {
// A new cache generation is created by each call to Commit.
// cachelimit sets the number of past cache generations to keep.
func NewSecure(root common.Hash, db *Database, cachelimit uint16) (*SecureTrie, error) {
return NewSecureWithOwner(common.Hash{}, root, db, cachelimit)
}
// NewSecureWithOwner creates a trie with an existing root node from a backing
// database with an assigned owner for storage proximity and optional intermediate
// in-memory node pool.
//
// If root is the zero hash or the sha3 hash of an empty string, the
// trie is initially empty. Otherwise, New will panic if db is nil
// and returns MissingNodeError if the root node cannot be found.
//
// Accessing the trie loads nodes from the database or node pool on demand.
// Loaded nodes are kept around until their 'cache generation' expires.
// A new cache generation is created by each call to Commit.
// cachelimit sets the number of past cache generations to keep.
func NewSecureWithOwner(owner common.Hash, root common.Hash, db *Database, cachelimit uint16) (*SecureTrie, error) {
if db == nil {
panic("trie.NewSecure called without a database")
}
trie, err := New(root, db)
trie, err := NewWithOwner(owner, root, db)
if err != nil {
return nil, err
}
@ -183,7 +199,7 @@ func (t *SecureTrie) NodeIterator(start []byte) NodeIterator {
// The caller must not hold onto the return value because it will become
// invalid on the next call to hashKey or secKey.
func (t *SecureTrie) hashKey(key []byte) []byte {
h := newHasher(0, 0, nil)
h := newHasher(t.trie.owner, 0, 0, nil)
h.sha.Reset()
h.sha.Write(key)
buf := h.sha.Sum(t.hashKeyBuf[:0])

2
trie/sync.go
View file

@ -280,7 +280,7 @@ func (s *Sync) children(req *request, object node) ([]*request, error) {
// 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 {
if err := req.callback(common.Hash{}, node, req.hash); err != nil {
return nil, err
}
}

28
trie/trie.go
View file

@ -57,7 +57,7 @@ func CacheUnloads() int64 {
// LeafCallback is a callback type invoked when a trie operation reaches a leaf
// node. It's used by state sync and commit to allow handling external references
// between account and storage tries.
type LeafCallback func(leaf []byte, parent common.Hash) error
type LeafCallback func(owner common.Hash, leaf []byte, parent common.Hash) error
// Trie is a Merkle Patricia Trie.
// The zero value is an empty trie with no database.
@ -65,8 +65,9 @@ type LeafCallback func(leaf []byte, parent common.Hash) error
//
// Trie is not safe for concurrent use.
type Trie struct {
db *Database
root node
db *Database
root node
owner common.Hash
// Cache generation values.
// cachegen increases by one with each commit operation.
@ -93,11 +94,23 @@ func (t *Trie) newFlag() nodeFlag {
// New will panic if db is nil and returns a MissingNodeError if root does
// not exist in the database. Accessing the trie loads nodes from db on demand.
func New(root common.Hash, db *Database) (*Trie, error) {
return NewWithOwner(common.Hash{}, root, db)
}
// NewWithOwner creates a trie with an existing root node from db and an assigned
// owner for storage proximity.
//
// If root is the zero hash or the sha3 hash of an empty string, the
// trie is initially empty and does not require a database. Otherwise,
// New will panic if db is nil and returns a MissingNodeError if root does
// not exist in the database. Accessing the trie loads nodes from db on demand.
func NewWithOwner(owner common.Hash, root common.Hash, db *Database) (*Trie, error) {
if db == nil {
panic("trie.New called without a database")
}
trie := &Trie{
db: db,
db: db,
owner: owner,
}
if root != (common.Hash{}) && root != emptyRoot {
rootnode, err := trie.resolveHash(root[:], nil)
@ -431,9 +444,10 @@ func (t *Trie) resolveHash(n hashNode, prefix []byte) (node, error) {
cacheMissCounter.Inc(1)
hash := common.BytesToHash(n)
if node := t.db.node(hash, t.cachegen); node != nil {
if node := t.db.node(t.owner, hash, t.cachegen); node != nil {
return node, nil
}
log.Warn("Missing trie node", "owner", t.owner.Hex(), "hash", hash.Hex(), "path", fmt.Sprintf("%x", prefix))
return nil, &MissingNodeError{NodeHash: hash, Path: prefix}
}
@ -468,7 +482,7 @@ func (t *Trie) hashRoot(db *Database, onleaf LeafCallback) (node, node, error) {
if t.root == nil {
return hashNode(emptyRoot.Bytes()), nil, nil
}
h := newHasher(t.cachegen, t.cachelimit, onleaf)
h := newHasher(t.owner, t.cachegen, t.cachelimit, onleaf)
defer returnHasherToPool(h)
return h.hash(t.root, db, true)
return h.hash(nil, t.root, db, true)
}

25
vendor/github.com/karalabe/cookiejar/LICENSE generated vendored Executable file
View file

@ -0,0 +1,25 @@
Copyright (c) 2014 Péter Szilágyi. All rights reserved.
Redistribution and use in source and binary forms, with or without modification,
are permitted provided that the following conditions are met:
* Redistributions of source code must retain the above copyright notice,
this list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
Alternatively, the CookieJar toolbox may be used in accordance with the terms
and conditions contained in a signed written agreement between you and the
author(s).

141
vendor/github.com/karalabe/cookiejar/collections/deque/deque.go generated vendored Executable file
View file

@ -0,0 +1,141 @@
// CookieJar - A contestant's algorithm toolbox
// Copyright (c) 2013 Peter Szilagyi. All rights reserved.
//
// CookieJar is dual licensed: use of this source code is governed by a BSD
// license that can be found in the LICENSE file. Alternatively, the CookieJar
// toolbox may be used in accordance with the terms and conditions contained
// in a signed written agreement between you and the author(s).
// Package deque implements a double ended queue supporting arbitrary types
// (even a mixture).
//
// Internally it uses a dynamically growing circular slice of blocks, resulting
// in faster resizes than a simple dynamic array/slice would allow, yet less gc
// overhead.
package deque
// The size of a block of data
const blockSize = 4096
// Double ended queue data structure.
type Deque struct {
leftIdx int
leftOff int
rightIdx int
rightOff int
blocks [][]interface{}
left []interface{}
right []interface{}
}
// Creates a new, empty deque.
func New() *Deque {
result := new(Deque)
result.blocks = [][]interface{}{make([]interface{}, blockSize)}
result.right = result.blocks[0]
result.left = result.blocks[0]
return result
}
// Pushes a new element into the queue from the right, expanding it if necessary.
func (d *Deque) PushRight(data interface{}) {
d.right[d.rightOff] = data
d.rightOff++
if d.rightOff == blockSize {
d.rightOff = 0
d.rightIdx = (d.rightIdx + 1) % len(d.blocks)
// If we wrapped over to the left, insert a new block and update indices
if d.rightIdx == d.leftIdx {
buffer := make([][]interface{}, len(d.blocks)+1)
copy(buffer[:d.rightIdx], d.blocks[:d.rightIdx])
buffer[d.rightIdx] = make([]interface{}, blockSize)
copy(buffer[d.rightIdx+1:], d.blocks[d.rightIdx:])
d.blocks = buffer
d.leftIdx++
d.left = d.blocks[d.leftIdx]
}
d.right = d.blocks[d.rightIdx]
}
}
// Pops out an element from the queue from the right. Note, no bounds checking are done.
func (d *Deque) PopRight() (res interface{}) {
d.rightOff--
if d.rightOff < 0 {
d.rightOff = blockSize - 1
d.rightIdx = (d.rightIdx - 1 + len(d.blocks)) % len(d.blocks)
d.right = d.blocks[d.rightIdx]
}
res, d.right[d.rightOff] = d.right[d.rightOff], nil
return
}
// Returns the rightmost element from the deque. No bounds are checked.
func (d *Deque) Right() interface{} {
if d.rightOff > 0 {
return d.right[d.rightOff-1]
} else {
return d.blocks[(d.rightIdx-1+len(d.blocks))%len(d.blocks)][blockSize-1]
}
}
// Pushes a new element into the queue from the left, expanding it if necessary.
func (d *Deque) PushLeft(data interface{}) {
d.leftOff--
if d.leftOff < 0 {
d.leftOff = blockSize - 1
d.leftIdx = (d.leftIdx - 1 + len(d.blocks)) % len(d.blocks)
// If we wrapped over to the right, insert a new block and update indices
if d.leftIdx == d.rightIdx {
d.leftIdx++
buffer := make([][]interface{}, len(d.blocks)+1)
copy(buffer[:d.leftIdx], d.blocks[:d.leftIdx])
buffer[d.leftIdx] = make([]interface{}, blockSize)
copy(buffer[d.leftIdx+1:], d.blocks[d.leftIdx:])
d.blocks = buffer
}
d.left = d.blocks[d.leftIdx]
}
d.left[d.leftOff] = data
}
// Pops out an element from the queue from the left. Note, no bounds checking are done.
func (d *Deque) PopLeft() (res interface{}) {
res, d.left[d.leftOff] = d.left[d.leftOff], nil
d.leftOff++
if d.leftOff == blockSize {
d.leftOff = 0
d.leftIdx = (d.leftIdx + 1) % len(d.blocks)
d.left = d.blocks[d.leftIdx]
}
return
}
// Returns the leftmost element from the deque. No bounds are checked.
func (d *Deque) Left() interface{} {
return d.left[d.leftOff]
}
// Checks whether the queue is empty.
func (d *Deque) Empty() bool {
return d.leftIdx == d.rightIdx && d.leftOff == d.rightOff
}
// Returns the number of elements in the queue.
func (d *Deque) Size() int {
if d.rightIdx > d.leftIdx {
return (d.rightIdx-d.leftIdx)*blockSize - d.leftOff + d.rightOff
} else if d.rightIdx < d.leftIdx {
return (len(d.blocks)-d.leftIdx+d.rightIdx)*blockSize - d.leftOff + d.rightOff
} else {
return d.rightOff - d.leftOff
}
}
// Clears out the contents of the queue.
func (d *Deque) Reset() {
*d = *New()
}

6
vendor/vendor.json vendored
View file

@ -266,6 +266,12 @@
"revision": "975b5c4c7c21c0e3d2764200bf2aa8e34657ae6e",
"revisionTime": "2017-04-30T22:20:11Z"
},
{
"checksumSHA1": "SDHLlmY5ED3dBFr1HfHvdcNaHao=",
"path": "github.com/karalabe/cookiejar/collections/deque",
"revision": "8dcd6a7f4951f6ff3ee9cbb919a06d8925822e57",
"revisionTime": "2015-07-24T13:16:13Z"
},
{
"checksumSHA1": "6XsjAARQFvlW6dS15al0ibTFPOQ=",
"path": "github.com/karalabe/hid",