From 2f469fd597996cb2f4298691b33be493c8265560 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Fri, 22 Jun 2018 15:37:43 +0300 Subject: [PATCH] core, ethdb, trie: historical state pruning --- core/blockchain.go | 65 +- core/genesis.go | 1 + core/rawdb/accessors_state.go | 43 ++ core/rawdb/schema.go | 7 + core/state/database.go | 12 +- core/state/statedb.go | 13 +- core/state/sync.go | 2 +- eth/api_tracer.go | 8 +- les/handler.go | 6 +- light/postprocess.go | 4 +- light/trie.go | 4 +- trie/database.go | 599 ++++++++++++------ trie/hasher.go | 25 +- trie/iterator.go | 7 +- trie/node.go | 2 +- trie/proof.go | 6 +- trie/pruning.go | 518 +++++++++++++++ trie/secure_trie.go | 20 +- trie/sync.go | 2 +- trie/trie.go | 28 +- vendor/github.com/karalabe/cookiejar/LICENSE | 25 + .../cookiejar/collections/deque/deque.go | 141 +++++ vendor/vendor.json | 6 + 23 files changed, 1289 insertions(+), 255 deletions(-) create mode 100644 core/rawdb/accessors_state.go create mode 100644 trie/pruning.go create mode 100755 vendor/github.com/karalabe/cookiejar/LICENSE create mode 100755 vendor/github.com/karalabe/cookiejar/collections/deque/deque.go diff --git a/core/blockchain.go b/core/blockchain.go index 71e806e6e8..cc6ffa9398 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -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)) diff --git a/core/genesis.go b/core/genesis.go index 4aa129966f..2e8db504ae 100644 --- a/core/genesis.go +++ b/core/genesis.go @@ -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) diff --git a/core/rawdb/accessors_state.go b/core/rawdb/accessors_state.go new file mode 100644 index 0000000000..1374281bdb --- /dev/null +++ b/core/rawdb/accessors_state.go @@ -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 . + +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) + } +} diff --git a/core/rawdb/schema.go b/core/rawdb/schema.go index 87dbf94fc0..d1c66bd755 100644 --- a/core/rawdb/schema.go +++ b/core/rawdb/schema.go @@ -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()...) diff --git a/core/state/database.go b/core/state/database.go index ce085747a6..2c091e6e11 100644 --- a/core/state/database.go +++ b/core/state/database.go @@ -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. diff --git a/core/state/statedb.go b/core/state/statedb.go index 8ad25a5824..c80d30b454 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -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 }) diff --git a/core/state/sync.go b/core/state/sync.go index 5290411a3b..5d2666af4a 100644 --- a/core/state/sync.go +++ b/core/state/sync.go @@ -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 diff --git a/eth/api_tracer.go b/eth/api_tracer.go index a529ea118e..f1d9c88e54 100644 --- a/eth/api_tracer.go +++ b/eth/api_tracer.go @@ -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) } diff --git a/les/handler.go b/les/handler.go index 9efe7d9e12..59cd60cce0 100644 --- a/les/handler.go +++ b/les/handler.go @@ -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 { diff --git a/light/postprocess.go b/light/postprocess.go index 2030782b1b..207a0661b5 100644 --- a/light/postprocess.go +++ b/light/postprocess.go @@ -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, } diff --git a/light/trie.go b/light/trie.go index 27abb1dc28..299436d0a7 100644 --- a/light/trie.go +++ b/light/trie.go @@ -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 } diff --git a/trie/database.go b/trie/database.go index 73ba2e761b..3951f94176 100644 --- a/trie/database.go +++ b/trie/database.go @@ -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: ", 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) } } diff --git a/trie/hasher.go b/trie/hasher.go index 9d6756b6f4..77c44094bb 100644 --- a/trie/hasher.go +++ b/trie/hasher.go @@ -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) } } } diff --git a/trie/iterator.go b/trie/iterator.go index 77f1681665..51e2f1e3b2 100644 --- a/trie/iterator.go +++ b/trie/iterator.go @@ -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) diff --git a/trie/node.go b/trie/node.go index 1fafb7a538..1ec56b1e9b 100644 --- a/trie/node.go +++ b/trie/node.go @@ -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 } diff --git a/trie/proof.go b/trie/proof.go index 0f18dd26bd..83bdecaaaf 100644 --- a/trie/proof.go +++ b/trie/proof.go @@ -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. diff --git a/trie/pruning.go b/trie/pruning.go new file mode 100644 index 0000000000..c45ad0c9c7 --- /dev/null +++ b/trie/pruning.go @@ -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 . + +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 + } +} diff --git a/trie/secure_trie.go b/trie/secure_trie.go index 6a50cfd5a6..507dcca846 100644 --- a/trie/secure_trie.go +++ b/trie/secure_trie.go @@ -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]) diff --git a/trie/sync.go b/trie/sync.go index ef931f633b..ac1f2b29cc 100644 --- a/trie/sync.go +++ b/trie/sync.go @@ -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 } } diff --git a/trie/trie.go b/trie/trie.go index af424d4ac6..f32d5dd591 100644 --- a/trie/trie.go +++ b/trie/trie.go @@ -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) } diff --git a/vendor/github.com/karalabe/cookiejar/LICENSE b/vendor/github.com/karalabe/cookiejar/LICENSE new file mode 100755 index 0000000000..467d60878d --- /dev/null +++ b/vendor/github.com/karalabe/cookiejar/LICENSE @@ -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). diff --git a/vendor/github.com/karalabe/cookiejar/collections/deque/deque.go b/vendor/github.com/karalabe/cookiejar/collections/deque/deque.go new file mode 100755 index 0000000000..eb31dddf8f --- /dev/null +++ b/vendor/github.com/karalabe/cookiejar/collections/deque/deque.go @@ -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() +} diff --git a/vendor/vendor.json b/vendor/vendor.json index a7cd0821e5..1d5d9a8f53 100644 --- a/vendor/vendor.json +++ b/vendor/vendor.json @@ -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",