From 33cca2bbeacefd196906778f1ab07e0c1a59a957 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 11 Jan 2018 19:03:51 +0200 Subject: [PATCH] core, trie: intermediate mempool between trie and database --- accounts/abi/bind/backends/simulated.go | 6 +- cmd/evm/runner.go | 2 +- cmd/geth/chaincmd.go | 2 +- core/blockchain.go | 18 ++- core/chain_makers.go | 2 +- core/genesis.go | 2 +- core/state/database.go | 19 ++- core/state/managed_state_test.go | 3 +- core/state/state_test.go | 5 +- core/state/statedb_test.go | 14 +- core/state/sync_test.go | 8 +- core/tx_pool_test.go | 32 ++--- core/vm/runtime/runtime.go | 4 +- core/vm/runtime/runtime_test.go | 2 +- eth/api.go | 4 +- eth/api_test.go | 2 +- eth/api_tracer.go | 10 +- eth/handler_test.go | 2 +- les/handler.go | 14 +- les/handler_test.go | 2 +- les/odr_test.go | 4 +- light/odr_test.go | 6 +- light/postprocess.go | 4 +- light/trie.go | 4 +- light/trie_test.go | 2 +- tests/state_test_util.go | 2 +- trie/hasher.go | 99 +++++++++++--- trie/iterator_test.go | 12 +- trie/mempool.go | 174 ++++++++++++++++++++++++ trie/proof.go | 4 +- trie/secure_trie.go | 4 +- trie/secure_trie_test.go | 4 +- trie/sync_test.go | 10 +- trie/trie.go | 39 ++++-- trie/trie_test.go | 42 +++--- 35 files changed, 424 insertions(+), 139 deletions(-) create mode 100644 trie/mempool.go diff --git a/accounts/abi/bind/backends/simulated.go b/accounts/abi/bind/backends/simulated.go index 1803d3f236..f1b79331e0 100644 --- a/accounts/abi/bind/backends/simulated.go +++ b/accounts/abi/bind/backends/simulated.go @@ -103,7 +103,7 @@ func (b *SimulatedBackend) Rollback() { func (b *SimulatedBackend) rollback() { blocks, _ := core.GenerateChain(b.config, b.blockchain.CurrentBlock(), ethash.NewFaker(), b.database, 1, func(int, *core.BlockGen) {}) b.pendingBlock = blocks[0] - b.pendingState, _ = state.New(b.pendingBlock.Root(), state.NewDatabase(b.database)) + b.pendingState, _ = state.New(b.pendingBlock.Root(), state.NewDatabase(b.database, nil)) } // CodeAt returns the code associated with a certain account in the blockchain. @@ -310,7 +310,7 @@ func (b *SimulatedBackend) SendTransaction(ctx context.Context, tx *types.Transa block.AddTx(tx) }) b.pendingBlock = blocks[0] - b.pendingState, _ = state.New(b.pendingBlock.Root(), state.NewDatabase(b.database)) + b.pendingState, _ = state.New(b.pendingBlock.Root(), state.NewDatabase(b.database, nil)) return nil } @@ -387,7 +387,7 @@ func (b *SimulatedBackend) AdjustTime(adjustment time.Duration) error { block.OffsetTime(int64(adjustment.Seconds())) }) b.pendingBlock = blocks[0] - b.pendingState, _ = state.New(b.pendingBlock.Root(), state.NewDatabase(b.database)) + b.pendingState, _ = state.New(b.pendingBlock.Root(), state.NewDatabase(b.database, nil)) return nil } diff --git a/cmd/evm/runner.go b/cmd/evm/runner.go index 96de0c76ac..5c792f09f0 100644 --- a/cmd/evm/runner.go +++ b/cmd/evm/runner.go @@ -100,7 +100,7 @@ func runCmd(ctx *cli.Context) error { chainConfig = gen.Config } else { db, _ := ethdb.NewMemDatabase() - statedb, _ = state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ = state.New(common.Hash{}, state.NewDatabase(db, nil)) } if ctx.GlobalString(SenderFlag.Name) != "" { sender = common.HexToAddress(ctx.GlobalString(SenderFlag.Name)) diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index 4a9a7b11b5..83bb611757 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -379,7 +379,7 @@ func dump(ctx *cli.Context) error { fmt.Println("{}") utils.Fatalf("block not found") } else { - state, err := state.New(block.Root(), state.NewDatabase(chainDb)) + state, err := state.New(block.Root(), state.NewDatabase(chainDb, nil)) if err != nil { utils.Fatalf("could not create new state: %v", err) } diff --git a/core/blockchain.go b/core/blockchain.go index d5e139e311..3dedff6375 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -96,6 +96,7 @@ type BlockChain struct { currentBlock *types.Block // Current head of the block chain currentFastBlock *types.Block // Current head of the fast-sync chain (may be above the block chain!) + trieMemPool *trie.MemPool // Trie node memory pool to avoid storing everything to disk stateCache state.Database // State database to reuse between imports (contains state cache) bodyCache *lru.Cache // Cache for the most recent block bodies bodyRLPCache *lru.Cache // Cache for the most recent block bodies in RLP encoded format @@ -120,6 +121,7 @@ type BlockChain struct { // available in the database. It initialises the default Ethereum Validator and // Processor. func NewBlockChain(chainDb ethdb.Database, config *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config) (*BlockChain, error) { + trieMemPool := trie.NewMemPool() bodyCache, _ := lru.New(bodyCacheLimit) bodyRLPCache, _ := lru.New(bodyCacheLimit) blockCache, _ := lru.New(blockCacheLimit) @@ -129,7 +131,8 @@ func NewBlockChain(chainDb ethdb.Database, config *params.ChainConfig, engine co bc := &BlockChain{ config: config, chainDb: chainDb, - stateCache: state.NewDatabase(chainDb), + trieMemPool: trieMemPool, + stateCache: state.NewDatabase(chainDb, trieMemPool), quit: make(chan struct{}), bodyCache: bodyCache, bodyRLPCache: bodyRLPCache, @@ -292,7 +295,7 @@ func (bc *BlockChain) FastSyncCommitHead(hash common.Hash) error { if block == nil { return fmt.Errorf("non existent block [%x…]", hash[:4]) } - if _, err := trie.NewSecure(block.Root(), bc.chainDb, 0); err != nil { + if _, err := trie.NewSecure(block.Root(), bc.chainDb, nil, 0); err != nil { return err } // If all checks out, manually set the head block @@ -791,9 +794,18 @@ func (bc *BlockChain) WriteBlockAndState(block *types.Block, receipts []*types.R if err := WriteBlock(batch, block); err != nil { return NonStatTy, err } - if _, err := state.CommitTo(batch, bc.config.IsEIP158(block.Number())); err != nil { + root, err := state.CommitTo(batch, bc.config.IsEIP158(block.Number())) + if err != nil { return NonStatTy, err } + bc.trieMemPool.Reference(root, common.Hash{}) + if number := block.NumberU64(); number > 192 { + if (number-192)%128 == 0 { + bc.trieMemPool.Commit(root, batch) + } + header := bc.GetHeaderByNumber(block.NumberU64() - 192) + bc.trieMemPool.Dereference(header.Root, common.Hash{}) + } if err := WriteBlockReceipts(batch, block.Hash(), block.NumberU64(), receipts); err != nil { return NonStatTy, err } diff --git a/core/chain_makers.go b/core/chain_makers.go index 5e264a9942..fd598abd97 100644 --- a/core/chain_makers.go +++ b/core/chain_makers.go @@ -201,7 +201,7 @@ func GenerateChain(config *params.ChainConfig, parent *types.Block, engine conse return nil, nil } for i := 0; i < n; i++ { - statedb, err := state.New(parent.Root(), state.NewDatabase(db)) + statedb, err := state.New(parent.Root(), state.NewDatabase(db, nil)) if err != nil { panic(err) } diff --git a/core/genesis.go b/core/genesis.go index e22985b800..3e7751cc0b 100644 --- a/core/genesis.go +++ b/core/genesis.go @@ -223,7 +223,7 @@ func (g *Genesis) configOrDefault(ghash common.Hash) *params.ChainConfig { // ToBlock creates the block and state of a genesis specification. func (g *Genesis) ToBlock() (*types.Block, *state.StateDB) { db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) for addr, account := range g.Alloc { statedb.AddBalance(addr, account.Balance) statedb.SetCode(addr, account.Code) diff --git a/core/state/database.go b/core/state/database.go index 946625e76e..0ddc7e164a 100644 --- a/core/state/database.go +++ b/core/state/database.go @@ -40,14 +40,18 @@ const ( // Database wraps access to tries and contract code. type Database interface { - // Accessing tries: // OpenTrie opens the main account trie. - // OpenStorageTrie opens the storage trie of an account. OpenTrie(root common.Hash) (Trie, error) + + // OpenStorageTrie opens the storage trie of an account. OpenStorageTrie(addrHash, root common.Hash) (Trie, error) - // Accessing contract code: + + // ContractCode retrieves a particular contract's code. ContractCode(addrHash, codeHash common.Hash) ([]byte, error) + + // ContractCode retrieves a particular contracts code's size. ContractCodeSize(addrHash, codeHash common.Hash) (int, error) + // CopyTrie returns an independent copy of the given trie. CopyTrie(Trie) Trie } @@ -65,13 +69,14 @@ type Trie interface { // NewDatabase creates a backing store for state. The returned database is safe for // concurrent use and retains cached trie nodes in memory. -func NewDatabase(db ethdb.Database) Database { +func NewDatabase(db ethdb.Database, pool *trie.MemPool) Database { csc, _ := lru.New(codeSizeCacheSize) - return &cachingDB{db: db, codeSizeCache: csc} + return &cachingDB{db: db, pool: pool, codeSizeCache: csc} } type cachingDB struct { db ethdb.Database + pool *trie.MemPool mu sync.Mutex pastTries []*trie.SecureTrie codeSizeCache *lru.Cache @@ -86,7 +91,7 @@ func (db *cachingDB) OpenTrie(root common.Hash) (Trie, error) { return cachedTrie{db.pastTries[i].Copy(), db}, nil } } - tr, err := trie.NewSecure(root, db.db, MaxTrieCacheGen) + tr, err := trie.NewSecure(root, db.db, db.pool, MaxTrieCacheGen) if err != nil { return nil, err } @@ -106,7 +111,7 @@ func (db *cachingDB) pushTrie(t *trie.SecureTrie) { } func (db *cachingDB) OpenStorageTrie(addrHash, root common.Hash) (Trie, error) { - return trie.NewSecure(root, db.db, 0) + return trie.NewSecure(root, db.db, db.pool, 0) } func (db *cachingDB) CopyTrie(t Trie) Trie { diff --git a/core/state/managed_state_test.go b/core/state/managed_state_test.go index 1cfdd3a890..f565164a40 100644 --- a/core/state/managed_state_test.go +++ b/core/state/managed_state_test.go @@ -21,13 +21,14 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/trie" ) var addr = common.BytesToAddress([]byte("test")) func create() (*ManagedState, *account) { db, _ := ethdb.NewMemDatabase() - statedb, _ := New(common.Hash{}, NewDatabase(db)) + statedb, _ := New(common.Hash{}, NewDatabase(db, trie.NewMemPool())) ms := ManageState(statedb) ms.StateDB.SetNonce(addr, 100) ms.accounts[addr] = newAccount(ms.StateDB.getStateObject(addr)) diff --git a/core/state/state_test.go b/core/state/state_test.go index bbae3685bb..bcfa1bbe6c 100644 --- a/core/state/state_test.go +++ b/core/state/state_test.go @@ -24,6 +24,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/trie" checker "gopkg.in/check.v1" ) @@ -88,7 +89,7 @@ func (s *StateSuite) TestDump(c *checker.C) { func (s *StateSuite) SetUpTest(c *checker.C) { s.db, _ = ethdb.NewMemDatabase() - s.state, _ = New(common.Hash{}, NewDatabase(s.db)) + s.state, _ = New(common.Hash{}, NewDatabase(s.db, trie.NewMemPool())) } func (s *StateSuite) TestNull(c *checker.C) { @@ -134,7 +135,7 @@ func (s *StateSuite) TestSnapshotEmpty(c *checker.C) { // printing/logging in tests (-check.vv does not work) func TestSnapshot2(t *testing.T) { db, _ := ethdb.NewMemDatabase() - state, _ := New(common.Hash{}, NewDatabase(db)) + state, _ := New(common.Hash{}, NewDatabase(db, trie.NewMemPool())) stateobjaddr0 := toAddr([]byte("so0")) stateobjaddr1 := toAddr([]byte("so1")) diff --git a/core/state/statedb_test.go b/core/state/statedb_test.go index 5c80e3aa56..02f4b872e9 100644 --- a/core/state/statedb_test.go +++ b/core/state/statedb_test.go @@ -33,6 +33,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/trie" ) // Tests that updating a state trie does not leak any database writes prior to @@ -40,7 +41,7 @@ import ( func TestUpdateLeaks(t *testing.T) { // Create an empty state database db, _ := ethdb.NewMemDatabase() - state, _ := New(common.Hash{}, NewDatabase(db)) + state, _ := New(common.Hash{}, NewDatabase(db, trie.NewMemPool())) // Update it with some accounts for i := byte(0); i < 255; i++ { @@ -68,8 +69,8 @@ func TestIntermediateLeaks(t *testing.T) { // Create two state databases, one transitioning to the final state, the other final from the beginning transDb, _ := ethdb.NewMemDatabase() finalDb, _ := ethdb.NewMemDatabase() - transState, _ := New(common.Hash{}, NewDatabase(transDb)) - finalState, _ := New(common.Hash{}, NewDatabase(finalDb)) + transState, _ := New(common.Hash{}, NewDatabase(transDb, trie.NewMemPool())) + finalState, _ := New(common.Hash{}, NewDatabase(finalDb, trie.NewMemPool())) modify := func(state *StateDB, addr common.Address, i, tweak byte) { state.SetBalance(addr, big.NewInt(int64(11*i)+int64(tweak))) @@ -123,7 +124,7 @@ func TestIntermediateLeaks(t *testing.T) { func TestCopy(t *testing.T) { // Create a random state test to copy and modify "independently" mem, _ := ethdb.NewMemDatabase() - orig, _ := New(common.Hash{}, NewDatabase(mem)) + orig, _ := New(common.Hash{}, NewDatabase(mem, trie.NewMemPool())) for i := byte(0); i < 255; i++ { obj := orig.GetOrNewStateObject(common.BytesToAddress([]byte{i})) @@ -335,7 +336,8 @@ func (test *snapshotTest) run() bool { // Run all actions and create snapshots. var ( db, _ = ethdb.NewMemDatabase() - state, _ = New(common.Hash{}, NewDatabase(db)) + mp = trie.NewMemPool() + state, _ = New(common.Hash{}, NewDatabase(db, trie.NewMemPool())) snapshotRevs = make([]int, len(test.snapshots)) sindex = 0 ) @@ -350,7 +352,7 @@ func (test *snapshotTest) run() bool { // Revert all snapshots in reverse order. Each revert must yield a state // that is equivalent to fresh state with all actions up the snapshot applied. for sindex--; sindex >= 0; sindex-- { - checkstate, _ := New(common.Hash{}, NewDatabase(db)) + checkstate, _ := New(common.Hash{}, NewDatabase(db, mp)) for _, action := range test.actions[:test.snapshots[sindex]] { action.fn(action, checkstate) } diff --git a/core/state/sync_test.go b/core/state/sync_test.go index 06c572ea62..168658a878 100644 --- a/core/state/sync_test.go +++ b/core/state/sync_test.go @@ -39,7 +39,7 @@ type testAccount struct { func makeTestState() (Database, *ethdb.MemDatabase, common.Hash, []*testAccount) { // Create an empty state mem, _ := ethdb.NewMemDatabase() - db := NewDatabase(mem) + db := NewDatabase(mem, trie.NewMemPool()) state, _ := New(common.Hash{}, db) // Fill it with some arbitrary data @@ -71,7 +71,7 @@ func makeTestState() (Database, *ethdb.MemDatabase, common.Hash, []*testAccount) // account array. func checkStateAccounts(t *testing.T, db ethdb.Database, root common.Hash, accounts []*testAccount) { // Check root availability and state contents - state, err := New(root, NewDatabase(db)) + state, err := New(root, NewDatabase(db, trie.NewMemPool())) if err != nil { t.Fatalf("failed to create state trie at %x: %v", root, err) } @@ -96,7 +96,7 @@ func checkTrieConsistency(db ethdb.Database, root common.Hash) error { if v, _ := db.Get(root[:]); v == nil { return nil // Consider a non existent state consistent. } - trie, err := trie.New(root, db) + trie, err := trie.New(root, db, trie.NewMemPool()) if err != nil { return err } @@ -112,7 +112,7 @@ func checkStateConsistency(db ethdb.Database, root common.Hash) error { if _, err := db.Get(root.Bytes()); err != nil { return nil // Consider a non existent state consistent. } - state, err := New(root, NewDatabase(db)) + state, err := New(root, NewDatabase(db, trie.NewMemPool())) if err != nil { return err } diff --git a/core/tx_pool_test.go b/core/tx_pool_test.go index cd11f2ba29..746a4f2132 100644 --- a/core/tx_pool_test.go +++ b/core/tx_pool_test.go @@ -79,7 +79,7 @@ func pricedTransaction(nonce uint64, gaslimit uint64, gasprice *big.Int, key *ec func setupTxPool() (*TxPool, *ecdsa.PrivateKey) { db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} key, _ := crypto.GenerateKey() @@ -159,7 +159,7 @@ func (c *testChain) State() (*state.StateDB, error) { stdb := c.statedb if *c.trigger { db, _ := ethdb.NewMemDatabase() - c.statedb, _ = state.New(common.Hash{}, state.NewDatabase(db)) + c.statedb, _ = state.New(common.Hash{}, state.NewDatabase(db, nil)) // simulate that the new head block included tx0 and tx1 c.statedb.SetNonce(c.address, 2) c.statedb.SetBalance(c.address, new(big.Int).SetUint64(params.Ether)) @@ -178,7 +178,7 @@ func TestStateChangeDuringTransactionPoolReset(t *testing.T) { db, _ = ethdb.NewMemDatabase() key, _ = crypto.GenerateKey() address = crypto.PubkeyToAddress(key.PublicKey) - statedb, _ = state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ = state.New(common.Hash{}, state.NewDatabase(db, nil)) trigger = false ) @@ -338,7 +338,7 @@ func TestTransactionChainFork(t *testing.T) { addr := crypto.PubkeyToAddress(key.PublicKey) resetState := func() { db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) statedb.AddBalance(addr, big.NewInt(100000000000000)) pool.chain = &testBlockChain{statedb, 1000000, new(event.Feed)} @@ -368,7 +368,7 @@ func TestTransactionDoubleNonce(t *testing.T) { addr := crypto.PubkeyToAddress(key.PublicKey) resetState := func() { db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) statedb.AddBalance(addr, big.NewInt(100000000000000)) pool.chain = &testBlockChain{statedb, 1000000, new(event.Feed)} @@ -737,7 +737,7 @@ func testTransactionQueueGlobalLimiting(t *testing.T, nolocals bool) { // Create the pool to test the limit enforcement with db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} config := testTxPoolConfig @@ -826,7 +826,7 @@ func testTransactionQueueTimeLimiting(t *testing.T, nolocals bool) { // Create the pool to test the non-expiration enforcement db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} config := testTxPoolConfig @@ -981,7 +981,7 @@ func TestTransactionPendingGlobalLimiting(t *testing.T) { // Create the pool to test the limit enforcement with db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} config := testTxPoolConfig @@ -1028,7 +1028,7 @@ func TestTransactionCapClearsFromAll(t *testing.T) { // Create the pool to test the limit enforcement with db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} config := testTxPoolConfig @@ -1063,7 +1063,7 @@ func TestTransactionPendingMinimumAllowance(t *testing.T) { // Create the pool to test the limit enforcement with db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} config := testTxPoolConfig @@ -1112,7 +1112,7 @@ func TestTransactionPoolRepricing(t *testing.T) { // Create the pool to test the pricing enforcement with db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} pool := NewTxPool(testTxPoolConfig, params.TestChainConfig, blockchain) @@ -1211,7 +1211,7 @@ func TestTransactionPoolRepricingKeepsLocals(t *testing.T) { // Create the pool to test the pricing enforcement with db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} pool := NewTxPool(testTxPoolConfig, params.TestChainConfig, blockchain) @@ -1274,7 +1274,7 @@ func TestTransactionPoolUnderpricing(t *testing.T) { // Create the pool to test the pricing enforcement with db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} config := testTxPoolConfig @@ -1376,7 +1376,7 @@ func TestTransactionReplacement(t *testing.T) { // Create the pool to test the pricing enforcement with db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} pool := NewTxPool(testTxPoolConfig, params.TestChainConfig, blockchain) @@ -1471,7 +1471,7 @@ func testTransactionJournaling(t *testing.T, nolocals bool) { // Create the original pool to inject transaction into the journal db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} config := testTxPoolConfig @@ -1570,7 +1570,7 @@ func TestTransactionStatusCheck(t *testing.T) { // Create the pool to test the status retrievals with db, _ := ethdb.NewMemDatabase() - statedb, _ := state.New(common.Hash{}, state.NewDatabase(db)) + statedb, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) blockchain := &testBlockChain{statedb, 1000000, new(event.Feed)} pool := NewTxPool(testTxPoolConfig, params.TestChainConfig, blockchain) diff --git a/core/vm/runtime/runtime.go b/core/vm/runtime/runtime.go index edbf541766..37fda55c9b 100644 --- a/core/vm/runtime/runtime.go +++ b/core/vm/runtime/runtime.go @@ -102,7 +102,7 @@ func Execute(code, input []byte, cfg *Config) ([]byte, *state.StateDB, error) { if cfg.State == nil { db, _ := ethdb.NewMemDatabase() - cfg.State, _ = state.New(common.Hash{}, state.NewDatabase(db)) + cfg.State, _ = state.New(common.Hash{}, state.NewDatabase(db, nil)) } var ( address = common.StringToAddress("contract") @@ -133,7 +133,7 @@ func Create(input []byte, cfg *Config) ([]byte, common.Address, uint64, error) { if cfg.State == nil { db, _ := ethdb.NewMemDatabase() - cfg.State, _ = state.New(common.Hash{}, state.NewDatabase(db)) + cfg.State, _ = state.New(common.Hash{}, state.NewDatabase(db, nil)) } var ( vmenv = NewEnv(cfg) diff --git a/core/vm/runtime/runtime_test.go b/core/vm/runtime/runtime_test.go index 2c4dc50265..f2b532cd88 100644 --- a/core/vm/runtime/runtime_test.go +++ b/core/vm/runtime/runtime_test.go @@ -95,7 +95,7 @@ func TestExecute(t *testing.T) { func TestCall(t *testing.T) { db, _ := ethdb.NewMemDatabase() - state, _ := state.New(common.Hash{}, state.NewDatabase(db)) + state, _ := state.New(common.Hash{}, state.NewDatabase(db, nil)) address := common.HexToAddress("0x0a") state.SetCode(address, []byte{ byte(vm.PUSH1), 10, diff --git a/eth/api.go b/eth/api.go index 0db3eb5548..ec30d058be 100644 --- a/eth/api.go +++ b/eth/api.go @@ -462,11 +462,11 @@ func (api *PrivateDebugAPI) getModifiedAccounts(startBlock, endBlock *types.Bloc return nil, fmt.Errorf("start block height (%d) must be less than end block height (%d)", startBlock.Number().Uint64(), endBlock.Number().Uint64()) } - oldTrie, err := trie.NewSecure(startBlock.Root(), api.eth.chainDb, 0) + oldTrie, err := trie.NewSecure(startBlock.Root(), api.eth.chainDb, nil, 0) if err != nil { return nil, err } - newTrie, err := trie.NewSecure(endBlock.Root(), api.eth.chainDb, 0) + newTrie, err := trie.NewSecure(endBlock.Root(), api.eth.chainDb, nil, 0) if err != nil { return nil, err } diff --git a/eth/api_test.go b/eth/api_test.go index 248bc3ab6a..b13162314f 100644 --- a/eth/api_test.go +++ b/eth/api_test.go @@ -32,7 +32,7 @@ func TestStorageRangeAt(t *testing.T) { // Create a state where account 0x010000... has a few storage entries. var ( db, _ = ethdb.NewMemDatabase() - state, _ = state.New(common.Hash{}, state.NewDatabase(db)) + state, _ = state.New(common.Hash{}, state.NewDatabase(db, nil)) addr = common.Address{0x01} keys = []common.Hash{ // hashes of Keys of storage common.HexToHash("340dd630ad21bf010b4e676dbfa9ba9a02175262d1fa356232cfde6cb5b47ef2"), diff --git a/eth/api_tracer.go b/eth/api_tracer.go index d49f077aed..57800d221f 100644 --- a/eth/api_tracer.go +++ b/eth/api_tracer.go @@ -200,7 +200,7 @@ func (api *PrivateDebugAPI) traceChain(ctx context.Context, start, end *types.Bl return nil, fmt.Errorf("parent block #%d not found", number-1) } } - statedb, err := state.New(start.Root(), state.NewDatabase(db)) + statedb, err := state.New(start.Root(), state.NewDatabase(db, nil)) if err != nil { // If the starting state is missing, allow some number of blocks to be reexecuted reexec := defaultTraceReexec @@ -213,7 +213,7 @@ func (api *PrivateDebugAPI) traceChain(ctx context.Context, start, end *types.Bl if start == nil { break } - if statedb, err = state.New(start.Root(), state.NewDatabase(db)); err == nil { + if statedb, err = state.New(start.Root(), state.NewDatabase(db, nil)); err == nil { break } } @@ -367,7 +367,7 @@ func (api *PrivateDebugAPI) traceChain(ctx context.Context, start, end *types.Bl db.Prune(root) log.Info("Pruned tracer state entries", "deleted", nodes-db.memdb.Len(), "left", db.memdb.Len(), "elapsed", time.Since(start)) - statedb, _ = state.New(root, state.NewDatabase(db)) + statedb, _ = state.New(root, state.NewDatabase(db, nil)) } } }() @@ -555,7 +555,7 @@ func (api *PrivateDebugAPI) computeStateDB(block *types.Block, reexec uint64) (* if block == nil { break } - if statedb, err = state.New(block.Root(), state.NewDatabase(db)); err == nil { + if statedb, err = state.New(block.Root(), state.NewDatabase(db, nil)); err == nil { break } } @@ -603,7 +603,7 @@ func (api *PrivateDebugAPI) computeStateDB(block *types.Block, reexec uint64) (* db.Prune(root) log.Info("Pruned tracer state entries", "deleted", nodes-db.memdb.Len(), "left", db.memdb.Len(), "elapsed", time.Since(begin)) - statedb, _ = state.New(root, state.NewDatabase(db)) + statedb, _ = state.New(root, state.NewDatabase(db, nil)) } } log.Info("Historical state regenerated", "block", block.NumberU64(), "elapsed", time.Since(start)) diff --git a/eth/handler_test.go b/eth/handler_test.go index 9a02eddfb2..1f6470b780 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -372,7 +372,7 @@ func testGetNodeData(t *testing.T, protocol int) { } accounts := []common.Address{testBank, acc1Addr, acc2Addr} for i := uint64(0); i <= pm.blockchain.CurrentBlock().NumberU64(); i++ { - trie, _ := state.New(pm.blockchain.GetBlockByNumber(i).Root(), state.NewDatabase(statedb)) + trie, _ := state.New(pm.blockchain.GetBlockByNumber(i).Root(), state.NewDatabase(statedb, nil)) for j, acc := range accounts { state, _ := pm.blockchain.State() diff --git a/les/handler.go b/les/handler.go index 8cd37c7abb..79146efa1b 100644 --- a/les/handler.go +++ b/les/handler.go @@ -579,7 +579,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { for _, req := range req.Reqs { // Retrieve the requested state entry, stopping if enough was found if header := core.GetHeader(pm.chainDb, req.BHash, core.GetBlockNumber(pm.chainDb, req.BHash)); header != nil { - if trie, _ := trie.New(header.Root, pm.chainDb); trie != nil { + if trie, _ := trie.New(header.Root, pm.chainDb, nil); trie != nil { sdata := trie.Get(req.AccKey) var acc state.Account if err := rlp.DecodeBytes(sdata, &acc); err == nil { @@ -706,13 +706,13 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } // Retrieve the requested state entry, stopping if enough was found if header := core.GetHeader(pm.chainDb, req.BHash, core.GetBlockNumber(pm.chainDb, req.BHash)); header != nil { - if tr, _ := trie.New(header.Root, pm.chainDb); tr != nil { + if tr, _ := trie.New(header.Root, pm.chainDb, nil); tr != nil { if len(req.AccKey) > 0 { sdata := tr.Get(req.AccKey) tr = nil var acc state.Account if err := rlp.DecodeBytes(sdata, &acc); err == nil { - tr, _ = trie.New(acc.Root, pm.chainDb) + tr, _ = trie.New(acc.Root, pm.chainDb, nil) } } if tr != nil { @@ -757,7 +757,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } if tr == nil || req.BHash != lastBHash { if header := core.GetHeader(pm.chainDb, req.BHash, core.GetBlockNumber(pm.chainDb, req.BHash)); header != nil { - tr, _ = trie.New(header.Root, pm.chainDb) + tr, _ = trie.New(header.Root, pm.chainDb, nil) } else { tr = nil } @@ -771,7 +771,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { str = nil var acc state.Account if err := rlp.DecodeBytes(sdata, &acc); err == nil { - str, _ = trie.New(acc.Root, pm.chainDb) + str, _ = trie.New(acc.Root, pm.chainDb, nil) } lastAccKey = common.CopyBytes(req.AccKey) } @@ -858,7 +858,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if header := pm.blockchain.GetHeaderByNumber(req.BlockNum); header != nil { sectionHead := core.GetCanonicalHash(pm.chainDb, req.ChtNum*light.ChtV1Frequency-1) if root := light.GetChtRoot(pm.chainDb, req.ChtNum-1, sectionHead); root != (common.Hash{}) { - if tr, _ := trie.New(root, trieDb); tr != nil { + if tr, _ := trie.New(root, trieDb, nil); tr != nil { var encNumber [8]byte binary.BigEndian.PutUint64(encNumber[:], req.BlockNum) var proof light.NodeList @@ -910,7 +910,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { var prefix string root, prefix = pm.getHelperTrie(req.HelperTrieType, req.TrieIdx) if root != (common.Hash{}) { - if t, err := trie.New(root, ethdb.NewTable(pm.chainDb, prefix)); err == nil { + if t, err := trie.New(root, ethdb.NewTable(pm.chainDb, prefix), nil); err == nil { tr = t } } diff --git a/les/handler_test.go b/les/handler_test.go index 10e5499a33..e0d775150c 100644 --- a/les/handler_test.go +++ b/les/handler_test.go @@ -359,7 +359,7 @@ func testGetProofs(t *testing.T, protocol int) { for i := uint64(0); i <= bc.CurrentBlock().NumberU64(); i++ { header := bc.GetHeaderByNumber(i) root := header.Root - trie, _ := trie.New(root, db) + trie, _ := trie.New(root, db, nil) for _, acc := range accounts { req := ProofReq{ diff --git a/les/odr_test.go b/les/odr_test.go index cf609be882..443752070b 100644 --- a/les/odr_test.go +++ b/les/odr_test.go @@ -90,7 +90,7 @@ func odrAccounts(ctx context.Context, db ethdb.Database, config *params.ChainCon for _, addr := range acc { if bc != nil { header := bc.GetHeaderByHash(bhash) - st, err = state.New(header.Root, state.NewDatabase(db)) + st, err = state.New(header.Root, state.NewDatabase(db, nil)) } else { header := lc.GetHeaderByHash(bhash) st = light.NewState(ctx, header, lc.Odr()) @@ -123,7 +123,7 @@ func odrContractCall(ctx context.Context, db ethdb.Database, config *params.Chai data[35] = byte(i) if bc != nil { header := bc.GetHeaderByHash(bhash) - statedb, err := state.New(header.Root, state.NewDatabase(db)) + statedb, err := state.New(header.Root, state.NewDatabase(db, nil)) if err == nil { from := statedb.GetOrNewStateObject(testBankAddress) diff --git a/light/odr_test.go b/light/odr_test.go index e3d07518a8..de715ef3d4 100644 --- a/light/odr_test.go +++ b/light/odr_test.go @@ -74,7 +74,7 @@ func (odr *testOdr) Retrieve(ctx context.Context, req OdrRequest) error { case *ReceiptsRequest: req.Receipts = core.GetBlockReceipts(odr.sdb, req.Hash, core.GetBlockNumber(odr.sdb, req.Hash)) case *TrieRequest: - t, _ := trie.New(req.Id.Root, odr.sdb) + t, _ := trie.New(req.Id.Root, odr.sdb, nil) nodes := NewNodeSet() t.Prove(req.Key, 0, nodes) req.Proof = nodes @@ -131,7 +131,7 @@ func odrAccounts(ctx context.Context, db ethdb.Database, bc *core.BlockChain, lc st = NewState(ctx, header, lc.Odr()) } else { header := bc.GetHeaderByHash(bhash) - st, _ = state.New(header.Root, state.NewDatabase(db)) + st, _ = state.New(header.Root, state.NewDatabase(db, nil)) } var res []byte @@ -171,7 +171,7 @@ func odrContractCall(ctx context.Context, db ethdb.Database, bc *core.BlockChain } else { chain = bc header = bc.GetHeaderByHash(bhash) - st, _ = state.New(header.Root, state.NewDatabase(db)) + st, _ = state.New(header.Root, state.NewDatabase(db, nil)) } // Perform read-only call. diff --git a/light/postprocess.go b/light/postprocess.go index 32dbc102be..7ec0a11ba1 100644 --- a/light/postprocess.go +++ b/light/postprocess.go @@ -141,7 +141,7 @@ func (c *ChtIndexerBackend) Reset(section uint64, lastSectionHead common.Hash) e root = GetChtRoot(c.db, section-1, lastSectionHead) } var err error - c.trie, err = trie.New(root, c.cdb) + c.trie, err = trie.New(root, c.cdb, nil) c.section = section return err } @@ -236,7 +236,7 @@ func (b *BloomTrieIndexerBackend) Reset(section uint64, lastSectionHead common.H root = GetBloomTrieRoot(b.db, section-1, lastSectionHead) } var err error - b.trie, err = trie.New(root, b.cdb) + b.trie, err = trie.New(root, b.cdb, nil) b.section = section return err } diff --git a/light/trie.go b/light/trie.go index 7a9c86b98d..cec1e1a3c8 100644 --- a/light/trie.go +++ b/light/trie.go @@ -141,7 +141,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, t.db.backend.Database()) + t.trie, err = trie.New(t.id.Root, t.db.backend.Database(), nil) } if err == nil { err = fn() @@ -167,7 +167,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, t.db.backend.Database()) + t, err := trie.New(t.id.Root, t.db.backend.Database(), nil) if err == nil { it.t.trie = t } diff --git a/light/trie_test.go b/light/trie_test.go index d99664718f..284b3fb748 100644 --- a/light/trie_test.go +++ b/light/trie_test.go @@ -50,7 +50,7 @@ func TestNodeIterator(t *testing.T) { odr := &testOdr{sdb: fulldb, ldb: lightdb} head := blockchain.CurrentHeader() lightTrie, _ := NewStateDatabase(ctx, head, odr).OpenTrie(head.Root) - fullTrie, _ := state.NewDatabase(fulldb).OpenTrie(head.Root) + fullTrie, _ := state.NewDatabase(fulldb, nil).OpenTrie(head.Root) if err := diffTries(fullTrie, lightTrie); err != nil { t.Fatal(err) } diff --git a/tests/state_test_util.go b/tests/state_test_util.go index 78c05b0245..c86c8f7398 100644 --- a/tests/state_test_util.go +++ b/tests/state_test_util.go @@ -159,7 +159,7 @@ func (t *StateTest) gasLimit(subtest StateSubtest) uint64 { } func MakePreState(db ethdb.Database, accounts core.GenesisAlloc) *state.StateDB { - sdb := state.NewDatabase(db) + sdb := state.NewDatabase(db, nil) statedb, _ := state.New(common.Hash{}, sdb) for addr, a := range accounts { statedb.SetCode(addr, a.Code) diff --git a/trie/hasher.go b/trie/hasher.go index 4719aabf62..1f66757501 100644 --- a/trie/hasher.go +++ b/trie/hasher.go @@ -19,6 +19,7 @@ package trie import ( "bytes" "hash" + "math/big" "sync" "github.com/ethereum/go-ethereum/common" @@ -51,10 +52,10 @@ 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 DatabaseWriter, force bool) (node, node, error) { +func (h *hasher) hash(n node, pool *MemPool, 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 { + if pool == nil { return hash, n, nil } if n.canUnload(h.cachegen, h.cachelimit) { @@ -68,11 +69,11 @@ func (h *hasher) hash(n node, db DatabaseWriter, force bool) (node, node, error) } } // Trie not processed yet or needs storage, walk the children - collapsed, cached, err := h.hashChildren(n, db) + collapsed, cached, refs, err := h.hashChildren(n, pool) if err != nil { return hashNode{}, n, err } - hashed, err := h.store(collapsed, db, force) + hashed, refs, err := h.store(collapsed, refs, pool, force) if err != nil { return hashNode{}, n, err } @@ -83,12 +84,12 @@ func (h *hasher) hash(n node, db DatabaseWriter, force bool) (node, node, error) switch cn := cached.(type) { case *shortNode: cn.flags.hash = cachedHash - if db != nil { + if pool != nil { cn.flags.dirty = false } case *fullNode: cn.flags.hash = cachedHash - if db != nil { + if pool != nil { cn.flags.dirty = false } } @@ -98,7 +99,7 @@ func (h *hasher) hash(n node, db DatabaseWriter, 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 DatabaseWriter) (node, node, error) { +func (h *hasher) hashChildren(original node, pool *MemPool) (node, node, []common.Hash, error) { var err error switch n := original.(type) { @@ -109,15 +110,15 @@ func (h *hasher) hashChildren(original node, db DatabaseWriter) (node, node, err 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(n.Val, pool, false) if err != nil { - return original, original, err + return original, original, nil, err } } if collapsed.Val == nil { collapsed.Val = valueNode(nil) // Ensure that nil children are encoded as empty strings. } - return collapsed, cached, nil + return collapsed, cached, h.externals(collapsed.Val), nil case *fullNode: // Hash the full node's children, caching the newly hashed subtrees @@ -125,9 +126,9 @@ func (h *hasher) hashChildren(original node, db DatabaseWriter) (node, node, err 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(n.Children[i], pool, false) if err != nil { - return original, original, err + return original, original, nil, err } } else { collapsed.Children[i] = valueNode(nil) // Ensure that nil children are encoded as empty strings. @@ -137,18 +138,53 @@ func (h *hasher) hashChildren(original node, db DatabaseWriter) (node, node, err if collapsed.Children[16] == nil { collapsed.Children[16] = valueNode(nil) } - return collapsed, cached, nil + var refs []common.Hash + for i := 0; i < 16; i++ { + refs = append(refs, h.externals(collapsed.Children[i])...) + } + return collapsed, cached, refs, nil default: // Value and hash nodes don't have children so they're left as were - return n, original, nil + return n, original, h.externals(n), nil } } -func (h *hasher) store(n node, db DatabaseWriter, force bool) (node, error) { +// externals returns any external nodes referenced by a particular node. The only +// current case for it is when an account trie references its storage trie. +func (h *hasher) externals(n node) []common.Hash { + // Only value nodes can reference external nodes + val, ok := n.(valueNode) + if !ok { + return nil + } + // Account nodes have very specific sizes, discard anything else + // TODO(karalabe): Seriously? Dafuq man?! + if size := len(val); size < 70 || size > 102 { + return nil + } + // Only account nodes can reference external storage tries + var account struct { + Nonce uint64 + Balance *big.Int + Root common.Hash + CodeHash []byte + } + if err := rlp.DecodeBytes(val, &account); err != nil { + //fmt.Printf(".") + return nil + } + // Empty tries are not referenced + if account.Root == emptyState { + return nil + } + return []common.Hash{account.Root} +} + +func (h *hasher) store(n node, refs []common.Hash, pool *MemPool, force bool) (node, []common.Hash, error) { // Don't store hashes or empty nodes. if _, isHash := n.(hashNode); n == nil || isHash { - return n, nil + return n, refs, nil } // Generate the RLP encoding of the node h.tmp.Reset() @@ -157,7 +193,7 @@ func (h *hasher) store(n node, db DatabaseWriter, force bool) (node, error) { } if h.tmp.Len() < 32 && !force { - return n, nil // Nodes smaller than 32 bytes are stored inside their parent + return n, refs, nil // Nodes smaller than 32 bytes are stored inside their parent } // Larger nodes are replaced by their hash and stored in the database. hash, _ := n.cache() @@ -166,8 +202,31 @@ func (h *hasher) store(n node, db DatabaseWriter, force bool) (node, error) { h.sha.Write(h.tmp.Bytes()) hash = hashNode(h.sha.Sum(nil)) } - if db != nil { - return hash, db.Put(hash, h.tmp.Bytes()) + if pool != nil { + // We are pooling the trie nodes into an intermediate memory cache + pool.lock.Lock() + defer pool.lock.Unlock() + + hash := common.BytesToHash(hash) + pool.insert(hash, h.tmp.Bytes()) + + // Track all direct parent->child node references + switch n := n.(type) { + case *shortNode: + if child, ok := n.Val.(hashNode); ok { + pool.reference(common.BytesToHash(child), hash) + } + case *fullNode: + for i := 0; i < 16; i++ { + if child, ok := n.Children[i].(hashNode); ok { + pool.reference(common.BytesToHash(child), hash) + } + } + } + // Track external references from account->storage trie + for _, ext := range refs { + pool.reference(ext, hash) + } } - return hash, nil + return hash, nil, nil } diff --git a/trie/iterator_test.go b/trie/iterator_test.go index 4808d8b0c6..f925f27d4a 100644 --- a/trie/iterator_test.go +++ b/trie/iterator_test.go @@ -280,7 +280,9 @@ func TestIteratorNoDups(t *testing.T) { // This test checks that nodeIterator.Next can be retried after inserting missing trie nodes. func TestIteratorContinueAfterError(t *testing.T) { db, _ := ethdb.NewMemDatabase() - tr, _ := New(common.Hash{}, db) + mp := NewMemPool() + + tr, _ := New(common.Hash{}, db, mp) for _, val := range testdata1 { tr.Update([]byte(val.k), []byte(val.v)) } @@ -291,7 +293,7 @@ func TestIteratorContinueAfterError(t *testing.T) { for i := 0; i < 20; i++ { // Create trie that will load all nodes from DB. - tr, _ := New(tr.Hash(), db) + tr, _ := New(tr.Hash(), db, mp) // Remove a random node from the database. It can't be the root node // because that one is already loaded. @@ -331,7 +333,9 @@ func TestIteratorContinueAfterError(t *testing.T) { func TestIteratorContinueAfterSeekError(t *testing.T) { // Commit test trie to db, then remove the node containing "bars". db, _ := ethdb.NewMemDatabase() - ctr, _ := New(common.Hash{}, db) + mp := NewMemPool() + + ctr, _ := New(common.Hash{}, db, mp) for _, val := range testdata1 { ctr.Update([]byte(val.k), []byte(val.v)) } @@ -342,7 +346,7 @@ func TestIteratorContinueAfterSeekError(t *testing.T) { // Create a new iterator that seeks to "bars". Seeking can't proceed because // the node is missing. - tr, _ := New(root, db) + tr, _ := New(root, db, mp) it := tr.NodeIterator([]byte("bars")) missing, ok := it.Error().(*MissingNodeError) if !ok { diff --git a/trie/mempool.go b/trie/mempool.go new file mode 100644 index 0000000000..7cfb53d8b2 --- /dev/null +++ b/trie/mempool.go @@ -0,0 +1,174 @@ +// Copyright 2017 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 ( + "sync" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/log" +) + +// MemPool 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 MemPool struct { + cache map[common.Hash][]byte // Cached data blocks of the trie nodes + + parents map[common.Hash]int // Number of live nodes referencing a given one + children map[common.Hash]map[common.Hash]struct{} // Set of children referenced by given nodes + + 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 + + size common.StorageSize // Storage size of the memory pool + lock sync.RWMutex +} + +// NewMemPool creates a new memory pool to store ephemeral trie nodes before they +// are written out to disk or garbage collected. +func NewMemPool() *MemPool { + pool := &MemPool{ + cache: make(map[common.Hash][]byte), + parents: make(map[common.Hash]int), + children: make(map[common.Hash]map[common.Hash]struct{}), + } + pool.children[common.Hash{}] = make(map[common.Hash]struct{}) + return pool +} + +// insert writes a new trie node to the memory pool if it's yet unknown. The pool +// will make a copy of the slice. +// +// Note, this method assumes that the pool's lock is held! +func (pool *MemPool) insert(hash common.Hash, blob []byte) { + if _, ok := pool.cache[hash]; ok { + return + } + pool.cache[hash] = common.CopyBytes(blob) + pool.children[hash] = make(map[common.Hash]struct{}) + + pool.size += common.StorageSize(common.HashLength + len(blob)) +} + +// Fetch retrieves a cached trie node from memory, or returns nil if the pool +// does not have this particular piece of data. +func (pool *MemPool) Fetch(hash common.Hash) []byte { + pool.lock.RLock() + defer pool.lock.RUnlock() + + return pool.cache[hash] +} + +// Reference adds a new reference from parent to node. +func (pool *MemPool) Reference(node common.Hash, parent common.Hash) { + pool.lock.RLock() + defer pool.lock.RUnlock() + + pool.reference(node, parent) +} + +// reference is the private locked version of Reference. +func (pool *MemPool) reference(node common.Hash, parent common.Hash) { + // If the node does not exist, it's a node pulled from disk, skip + if _, ok := pool.cache[node]; !ok { + return + } + pool.parents[node]++ + pool.children[parent][node] = struct{}{} +} + +// Dereference removes an existing reference from parent to node. +func (pool *MemPool) Dereference(node common.Hash, parent common.Hash) { + pool.lock.Lock() + defer pool.lock.Unlock() + + nodes, storage, start := len(pool.cache), pool.size, time.Now() + pool.dereference(node, parent) + + pool.gcnodes += uint64(nodes - len(pool.cache)) + pool.gcsize += storage - pool.size + pool.gctime += time.Since(start) +} + +// dereference is the private locked version of Dereference. +func (pool *MemPool) dereference(node common.Hash, parent common.Hash) { + // If the node does not exist, it's a previously comitted node. + blob, ok := pool.cache[node] + if !ok { + return + } + delete(pool.children[parent], node) + pool.parents[node]-- + + // If there are no more references to the child, delete it and cascade + if pool.parents[node] == 0 { + for child := range pool.children[node] { + pool.dereference(child, node) + } + delete(pool.cache, node) + delete(pool.parents, node) + delete(pool.children, node) + + pool.size -= common.StorageSize(common.HashLength + len(blob)) + } +} + +// Commit iterates over all the children of a particular node, writes them out +// to disk, forcefully tearing down all references in both directions. +func (pool *MemPool) Commit(node common.Hash, db DatabaseWriter) { + pool.lock.Lock() + defer pool.lock.Unlock() + + nodes, storage, start := len(pool.cache), pool.size, time.Now() + pool.commit(node, db) + + log.Debug("Committed trie from memory pool", "nodes", nodes-len(pool.cache), "size", storage-pool.size, "time", time.Since(start), + "gcnodes", pool.gcnodes, "gcsize", pool.gcsize, "gctime", pool.gctime, "livenodes", len(pool.cache), "livesize", pool.size) + + // Reset the garbage collection statistics + pool.gcnodes, pool.gcsize, pool.gctime = 0, 0, 0 + + // Sanity check that we don't have dangling nodes in the pool (missing refs) + for hash, refs := range pool.parents { + if refs == 0 { + log.Warn("dangling node in mempool", "hash", hash) + break + } + } +} + +// commit is the private locked version of Commit. +func (pool *MemPool) commit(node common.Hash, db DatabaseWriter) { + // If the node does not exist, it's a previously comitted node. + blob, ok := pool.cache[node] + if !ok { + return + } + for child := range pool.children[node] { + pool.commit(child, db) + } + db.Put(node[:], blob) + + delete(pool.cache, node) + delete(pool.parents, node) + delete(pool.children, node) + + pool.size -= common.StorageSize(common.HashLength + len(blob)) +} diff --git a/trie/proof.go b/trie/proof.go index 5e886a2598..b9a2701bd8 100644 --- a/trie/proof.go +++ b/trie/proof.go @@ -70,8 +70,8 @@ func (t *Trie) Prove(key []byte, fromLevel uint, proofDb DatabaseWriter) error { 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(n, nil) + hn, _, _ := hasher.store(n, nil, 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/secure_trie.go b/trie/secure_trie.go index 20c303f31c..63d83fd5d6 100644 --- a/trie/secure_trie.go +++ b/trie/secure_trie.go @@ -55,11 +55,11 @@ type SecureTrie struct { // 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 NewSecure(root common.Hash, db Database, cachelimit uint16) (*SecureTrie, error) { +func NewSecure(root common.Hash, db Database, pool *MemPool, cachelimit uint16) (*SecureTrie, error) { if db == nil { panic("NewSecure called with nil database") } - trie, err := New(root, db) + trie, err := New(root, db, pool) if err != nil { return nil, err } diff --git a/trie/secure_trie_test.go b/trie/secure_trie_test.go index d74102e2a2..bef00fa78a 100644 --- a/trie/secure_trie_test.go +++ b/trie/secure_trie_test.go @@ -29,7 +29,7 @@ import ( func newEmptySecure() *SecureTrie { db, _ := ethdb.NewMemDatabase() - trie, _ := NewSecure(common.Hash{}, db, 0) + trie, _ := NewSecure(common.Hash{}, db, NewMemPool(), 0) return trie } @@ -37,7 +37,7 @@ func newEmptySecure() *SecureTrie { func makeTestSecureTrie() (ethdb.Database, *SecureTrie, map[string][]byte) { // Create an empty trie db, _ := ethdb.NewMemDatabase() - trie, _ := NewSecure(common.Hash{}, db, 0) + trie, _ := NewSecure(common.Hash{}, db, NewMemPool(), 0) // Fill it with some arbitrary data content := make(map[string][]byte) diff --git a/trie/sync_test.go b/trie/sync_test.go index ec16a25bd9..f1d3597863 100644 --- a/trie/sync_test.go +++ b/trie/sync_test.go @@ -28,7 +28,7 @@ import ( func makeTestTrie() (ethdb.Database, *Trie, map[string][]byte) { // Create an empty trie db, _ := ethdb.NewMemDatabase() - trie, _ := New(common.Hash{}, db) + trie, _ := New(common.Hash{}, db, nil) // Fill it with some arbitrary data content := make(map[string][]byte) @@ -59,7 +59,7 @@ func makeTestTrie() (ethdb.Database, *Trie, map[string][]byte) { // content map. func checkTrieContents(t *testing.T, db Database, root []byte, content map[string][]byte) { // Check root availability and trie contents - trie, err := New(common.BytesToHash(root), db) + trie, err := New(common.BytesToHash(root), db, nil) if err != nil { t.Fatalf("failed to create trie at %x: %v", root, err) } @@ -76,7 +76,7 @@ func checkTrieContents(t *testing.T, db Database, root []byte, content map[strin // checkTrieConsistency checks that all nodes in a trie are indeed present. func checkTrieConsistency(db Database, root common.Hash) error { // Create and iterate a trie rooted in a subnode - trie, err := New(root, db) + trie, err := New(root, db, nil) if err != nil { return nil // Consider a non existent state consistent } @@ -88,8 +88,8 @@ func checkTrieConsistency(db Database, root common.Hash) error { // Tests that an empty trie is not scheduled for syncing. func TestEmptyTrieSync(t *testing.T) { - emptyA, _ := New(common.Hash{}, nil) - emptyB, _ := New(emptyRoot, nil) + emptyA, _ := New(common.Hash{}, nil, nil) + emptyB, _ := New(emptyRoot, nil, nil) for i, trie := range []*Trie{emptyA, emptyB} { db, _ := ethdb.NewMemDatabase() diff --git a/trie/trie.go b/trie/trie.go index 8fe98d8351..07da38e82c 100644 --- a/trie/trie.go +++ b/trie/trie.go @@ -85,6 +85,7 @@ type DatabaseWriter interface { type Trie struct { root node db Database + pool *MemPool originalRoot common.Hash // Cache generation values. @@ -111,8 +112,8 @@ func (t *Trie) newFlag() nodeFlag { // 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 New(root common.Hash, db Database) (*Trie, error) { - trie := &Trie{db: db, originalRoot: root} +func New(root common.Hash, db Database, pool *MemPool) (*Trie, error) { + trie := &Trie{db: db, pool: pool, originalRoot: root} if (root != common.Hash{}) && root != emptyRoot { if db == nil { panic("trie.New: cannot use existing root without a database") @@ -447,12 +448,19 @@ func (t *Trie) resolve(n node, prefix []byte) (node, error) { func (t *Trie) resolveHash(n hashNode, prefix []byte) (node, error) { cacheMissCounter.Inc(1) + // Try to load the node from the recent mempool + hash := common.BytesToHash(n) + if t.pool != nil { + if enc := t.pool.Fetch(hash); enc != nil { + return mustDecodeNode(n, enc, t.cachegen), nil + } + } + // Node not in the mempool, load it from disk enc, err := t.db.Get(n) if err != nil || enc == nil { - return nil, &MissingNodeError{NodeHash: common.BytesToHash(n), Path: prefix} + return nil, &MissingNodeError{NodeHash: hash, Path: prefix} } - dec := mustDecodeNode(n, enc, t.cachegen) - return dec, nil + return mustDecodeNode(n, enc, t.cachegen), nil } // Root returns the root hash of the trie. @@ -487,7 +495,22 @@ func (t *Trie) Commit() (root common.Hash, err error) { // the changes made to db are written back to the trie's attached // database before using the trie. func (t *Trie) CommitTo(db DatabaseWriter) (root common.Hash, err error) { - hash, cached, err := t.hashRoot(db) + // Retrieve the intermedia trie node memory cache if really writing + var pool *MemPool + if db != nil { + if pool = t.pool; pool == nil { + // If the trie has no intermediate memory pool, but actual database write was + // nonetheless requested, store into an emphemeral pool and flush out to disk. + pool = NewMemPool() + defer func() { + for hash, blob := range pool.cache { + db.Put(hash[:], blob) + } + }() + } + } + // Calculate the root hash and store in the mempool if requested + hash, cached, err := t.hashRoot(pool) if err != nil { return common.Hash{}, err } @@ -496,11 +519,11 @@ func (t *Trie) CommitTo(db DatabaseWriter) (root common.Hash, err error) { return common.BytesToHash(hash.(hashNode)), nil } -func (t *Trie) hashRoot(db DatabaseWriter) (node, node, error) { +func (t *Trie) hashRoot(pool *MemPool) (node, node, error) { if t.root == nil { return hashNode(emptyRoot.Bytes()), nil, nil } h := newHasher(t.cachegen, t.cachelimit) defer returnHasherToPool(h) - return h.hash(t.root, db, true) + return h.hash(t.root, pool, true) } diff --git a/trie/trie_test.go b/trie/trie_test.go index 1e28c3bc48..5266adf242 100644 --- a/trie/trie_test.go +++ b/trie/trie_test.go @@ -44,7 +44,7 @@ func init() { // Used for testing func newEmpty() *Trie { db, _ := ethdb.NewMemDatabase() - trie, _ := New(common.Hash{}, db) + trie, _ := New(common.Hash{}, db, NewMemPool()) return trie } @@ -69,7 +69,7 @@ func TestNull(t *testing.T) { func TestMissingRoot(t *testing.T) { db, _ := ethdb.NewMemDatabase() - trie, err := New(common.HexToHash("0beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33"), db) + trie, err := New(common.HexToHash("0beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33"), db, NewMemPool()) if trie != nil { t.Error("New returned non-nil trie for invalid root") } @@ -80,36 +80,38 @@ func TestMissingRoot(t *testing.T) { func TestMissingNode(t *testing.T) { db, _ := ethdb.NewMemDatabase() - trie, _ := New(common.Hash{}, db) + mp := NewMemPool() + + trie, _ := New(common.Hash{}, db, mp) updateString(trie, "120000", "qwerqwerqwerqwerqwerqwerqwerqwer") updateString(trie, "123456", "asdfasdfasdfasdfasdfasdfasdfasdf") root, _ := trie.Commit() - trie, _ = New(root, db) + trie, _ = New(root, db, mp) _, err := trie.TryGet([]byte("120000")) if err != nil { t.Errorf("Unexpected error: %v", err) } - trie, _ = New(root, db) + trie, _ = New(root, db, mp) _, err = trie.TryGet([]byte("120099")) if err != nil { t.Errorf("Unexpected error: %v", err) } - trie, _ = New(root, db) + trie, _ = New(root, db, mp) _, err = trie.TryGet([]byte("123456")) if err != nil { t.Errorf("Unexpected error: %v", err) } - trie, _ = New(root, db) + trie, _ = New(root, db, mp) err = trie.TryUpdate([]byte("120099"), []byte("zxcvzxcvzxcvzxcvzxcvzxcvzxcvzxcv")) if err != nil { t.Errorf("Unexpected error: %v", err) } - trie, _ = New(root, db) + trie, _ = New(root, db, mp) err = trie.TryDelete([]byte("123456")) if err != nil { t.Errorf("Unexpected error: %v", err) @@ -117,31 +119,31 @@ func TestMissingNode(t *testing.T) { db.Delete(common.FromHex("e1d943cc8f061a0c0b98162830b970395ac9315654824bf21b73b891365262f9")) - trie, _ = New(root, db) + trie, _ = New(root, db, mp) _, err = trie.TryGet([]byte("120000")) if _, ok := err.(*MissingNodeError); !ok { t.Errorf("Wrong error: %v", err) } - trie, _ = New(root, db) + trie, _ = New(root, db, mp) _, err = trie.TryGet([]byte("120099")) if _, ok := err.(*MissingNodeError); !ok { t.Errorf("Wrong error: %v", err) } - trie, _ = New(root, db) + trie, _ = New(root, db, mp) _, err = trie.TryGet([]byte("123456")) if err != nil { t.Errorf("Unexpected error: %v", err) } - trie, _ = New(root, db) + trie, _ = New(root, db, mp) err = trie.TryUpdate([]byte("120099"), []byte("zxcv")) if _, ok := err.(*MissingNodeError); !ok { t.Errorf("Wrong error: %v", err) } - trie, _ = New(root, db) + trie, _ = New(root, db, mp) err = trie.TryDelete([]byte("123456")) if _, ok := err.(*MissingNodeError); !ok { t.Errorf("Wrong error: %v", err) @@ -269,7 +271,7 @@ func TestReplication(t *testing.T) { } // create a new trie on top of the database and check that lookups work. - trie2, err := New(exp, trie.db) + trie2, err := New(exp, trie.db, trie.pool) if err != nil { t.Fatalf("can't recreate trie at %x: %v", exp, err) } @@ -338,7 +340,7 @@ func TestCacheUnload(t *testing.T) { // The branch containing it is loaded from DB exactly two times: // in the 0th and 6th iteration. db := &countingDB{Database: trie.db, gets: make(map[string]int)} - trie, _ = New(root, db) + trie, _ = New(root, db, trie.pool) trie.SetCacheLimit(5) for i := 0; i < 12; i++ { getString(trie, key1) @@ -408,7 +410,9 @@ func (randTest) Generate(r *rand.Rand, size int) reflect.Value { func runRandTest(rt randTest) bool { db, _ := ethdb.NewMemDatabase() - tr, _ := New(common.Hash{}, db) + mp := NewMemPool() + + tr, _ := New(common.Hash{}, db, mp) values := make(map[string]string) // tracks content of the trie for i, step := range rt { @@ -435,14 +439,14 @@ func runRandTest(rt randTest) bool { rt[i].err = err return false } - newtr, err := New(hash, db) + newtr, err := New(hash, db, mp) if err != nil { rt[i].err = err return false } tr = newtr case opItercheckhash: - checktr, _ := New(common.Hash{}, nil) + checktr, _ := New(common.Hash{}, nil, nil) it := NewIterator(tr.NodeIterator(nil)) for it.Next() { checktr.Update(it.Key, it.Value) @@ -515,7 +519,7 @@ func benchGet(b *testing.B, commit bool) { trie := new(Trie) if commit { _, tmpdb := tempDB() - trie, _ = New(common.Hash{}, tmpdb) + trie, _ = New(common.Hash{}, tmpdb, nil) } k := make([]byte, 32) for i := 0; i < benchElemCount; i++ {