From 55599ee95d4151a2502465e0afc7c47bd1acba77 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 5 Feb 2018 18:40:32 +0200 Subject: core, trie: intermediate mempool between trie and database (#15857) This commit reduces database I/O by not writing every state trie to disk. --- light/lightchain.go | 7 ++++++ light/nodeset.go | 8 +++---- light/odr_test.go | 4 ++-- light/postprocess.go | 64 ++++++++++++++++++++++++++++------------------------ light/trie.go | 18 +++++++++++---- light/trie_test.go | 2 +- light/txpool_test.go | 2 +- 7 files changed, 64 insertions(+), 41 deletions(-) (limited to 'light') diff --git a/light/lightchain.go b/light/lightchain.go index f47957512..24529ef82 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -18,6 +18,7 @@ package light import ( "context" + "errors" "math/big" "sync" "sync/atomic" @@ -26,6 +27,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" @@ -212,6 +214,11 @@ func (bc *LightChain) Genesis() *types.Block { return bc.genesisBlock } +// State returns a new mutable state based on the current HEAD block. +func (bc *LightChain) State() (*state.StateDB, error) { + return nil, errors.New("not implemented, needs client/server interface split") +} + // GetBody retrieves a block body (transactions and uncles) from the database // or ODR service by hash, caching it if found. func (self *LightChain) GetBody(ctx context.Context, hash common.Hash) (*types.Body, error) { diff --git a/light/nodeset.go b/light/nodeset.go index c530a4fbe..ffdb71bb7 100644 --- a/light/nodeset.go +++ b/light/nodeset.go @@ -22,8 +22,8 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/rlp" - "github.com/ethereum/go-ethereum/trie" ) // NodeSet stores a set of trie nodes. It implements trie.Database and can also @@ -99,7 +99,7 @@ func (db *NodeSet) NodeList() NodeList { } // Store writes the contents of the set to the given database -func (db *NodeSet) Store(target trie.Database) { +func (db *NodeSet) Store(target ethdb.Putter) { db.lock.RLock() defer db.lock.RUnlock() @@ -108,11 +108,11 @@ func (db *NodeSet) Store(target trie.Database) { } } -// NodeList stores an ordered list of trie nodes. It implements trie.DatabaseWriter. +// NodeList stores an ordered list of trie nodes. It implements ethdb.Putter. type NodeList []rlp.RawValue // Store writes the contents of the list to the given database -func (n NodeList) Store(db trie.Database) { +func (n NodeList) Store(db ethdb.Putter) { for _, node := range n { db.Put(crypto.Keccak256(node), node) } diff --git a/light/odr_test.go b/light/odr_test.go index e3d07518a..d3f9374fd 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, trie.NewDatabase(odr.sdb)) nodes := NewNodeSet() t.Prove(req.Key, 0, nodes) req.Proof = nodes @@ -239,7 +239,7 @@ func testChainOdr(t *testing.T, protocol int, fn odrTestFn) { ) gspec.MustCommit(ldb) // Assemble the test environment - blockchain, _ := core.NewBlockChain(sdb, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}) + blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}) gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), sdb, 4, testChainGen) if _, err := blockchain.InsertChain(gchain); err != nil { t.Fatal(err) diff --git a/light/postprocess.go b/light/postprocess.go index 32dbc102b..bbac58d12 100644 --- a/light/postprocess.go +++ b/light/postprocess.go @@ -113,7 +113,8 @@ func StoreChtRoot(db ethdb.Database, sectionIdx uint64, sectionHead, root common // ChtIndexerBackend implements core.ChainIndexerBackend type ChtIndexerBackend struct { - db, cdb ethdb.Database + diskdb ethdb.Database + triedb *trie.Database section, sectionSize uint64 lastHash common.Hash trie *trie.Trie @@ -121,8 +122,6 @@ type ChtIndexerBackend struct { // NewBloomTrieIndexer creates a BloomTrie chain indexer func NewChtIndexer(db ethdb.Database, clientMode bool) *core.ChainIndexer { - cdb := ethdb.NewTable(db, ChtTablePrefix) - idb := ethdb.NewTable(db, "chtIndex-") var sectionSize, confirmReq uint64 if clientMode { sectionSize = ChtFrequency @@ -131,17 +130,23 @@ func NewChtIndexer(db ethdb.Database, clientMode bool) *core.ChainIndexer { sectionSize = ChtV1Frequency confirmReq = HelperTrieProcessConfirmations } - return core.NewChainIndexer(db, idb, &ChtIndexerBackend{db: db, cdb: cdb, sectionSize: sectionSize}, sectionSize, confirmReq, time.Millisecond*100, "cht") + idb := ethdb.NewTable(db, "chtIndex-") + backend := &ChtIndexerBackend{ + diskdb: db, + triedb: trie.NewDatabase(ethdb.NewTable(db, ChtTablePrefix)), + sectionSize: sectionSize, + } + return core.NewChainIndexer(db, idb, backend, sectionSize, confirmReq, time.Millisecond*100, "cht") } // Reset implements core.ChainIndexerBackend func (c *ChtIndexerBackend) Reset(section uint64, lastSectionHead common.Hash) error { var root common.Hash if section > 0 { - root = GetChtRoot(c.db, section-1, lastSectionHead) + root = GetChtRoot(c.diskdb, section-1, lastSectionHead) } var err error - c.trie, err = trie.New(root, c.cdb) + c.trie, err = trie.New(root, c.triedb) c.section = section return err } @@ -151,7 +156,7 @@ func (c *ChtIndexerBackend) Process(header *types.Header) { hash, num := header.Hash(), header.Number.Uint64() c.lastHash = hash - td := core.GetTd(c.db, hash, num) + td := core.GetTd(c.diskdb, hash, num) if td == nil { panic(nil) } @@ -163,17 +168,16 @@ func (c *ChtIndexerBackend) Process(header *types.Header) { // Commit implements core.ChainIndexerBackend func (c *ChtIndexerBackend) Commit() error { - batch := c.cdb.NewBatch() - root, err := c.trie.CommitTo(batch) + root, err := c.trie.Commit(nil) if err != nil { return err - } else { - batch.Write() - if ((c.section+1)*c.sectionSize)%ChtFrequency == 0 { - log.Info("Storing CHT", "idx", c.section*c.sectionSize/ChtFrequency, "sectionHead", fmt.Sprintf("%064x", c.lastHash), "root", fmt.Sprintf("%064x", root)) - } - StoreChtRoot(c.db, c.section, c.lastHash, root) } + c.triedb.Commit(root, false) + + if ((c.section+1)*c.sectionSize)%ChtFrequency == 0 { + log.Info("Storing CHT", "idx", c.section*c.sectionSize/ChtFrequency, "sectionHead", fmt.Sprintf("%064x", c.lastHash), "root", fmt.Sprintf("%064x", root)) + } + StoreChtRoot(c.diskdb, c.section, c.lastHash, root) return nil } @@ -205,7 +209,8 @@ func StoreBloomTrieRoot(db ethdb.Database, sectionIdx uint64, sectionHead, root // BloomTrieIndexerBackend implements core.ChainIndexerBackend type BloomTrieIndexerBackend struct { - db, cdb ethdb.Database + diskdb ethdb.Database + triedb *trie.Database section, parentSectionSize, bloomTrieRatio uint64 trie *trie.Trie sectionHeads []common.Hash @@ -213,9 +218,12 @@ type BloomTrieIndexerBackend struct { // NewBloomTrieIndexer creates a BloomTrie chain indexer func NewBloomTrieIndexer(db ethdb.Database, clientMode bool) *core.ChainIndexer { - cdb := ethdb.NewTable(db, BloomTrieTablePrefix) + backend := &BloomTrieIndexerBackend{ + diskdb: db, + triedb: trie.NewDatabase(ethdb.NewTable(db, BloomTrieTablePrefix)), + } idb := ethdb.NewTable(db, "bltIndex-") - backend := &BloomTrieIndexerBackend{db: db, cdb: cdb} + var confirmReq uint64 if clientMode { backend.parentSectionSize = BloomTrieFrequency @@ -233,10 +241,10 @@ func NewBloomTrieIndexer(db ethdb.Database, clientMode bool) *core.ChainIndexer func (b *BloomTrieIndexerBackend) Reset(section uint64, lastSectionHead common.Hash) error { var root common.Hash if section > 0 { - root = GetBloomTrieRoot(b.db, section-1, lastSectionHead) + root = GetBloomTrieRoot(b.diskdb, section-1, lastSectionHead) } var err error - b.trie, err = trie.New(root, b.cdb) + b.trie, err = trie.New(root, b.triedb) b.section = section return err } @@ -259,7 +267,7 @@ func (b *BloomTrieIndexerBackend) Commit() error { binary.BigEndian.PutUint64(encKey[2:10], b.section) var decomp []byte for j := uint64(0); j < b.bloomTrieRatio; j++ { - data, err := core.GetBloomBits(b.db, i, b.section*b.bloomTrieRatio+j, b.sectionHeads[j]) + data, err := core.GetBloomBits(b.diskdb, i, b.section*b.bloomTrieRatio+j, b.sectionHeads[j]) if err != nil { return err } @@ -279,17 +287,15 @@ func (b *BloomTrieIndexerBackend) Commit() error { b.trie.Delete(encKey[:]) } } - - batch := b.cdb.NewBatch() - root, err := b.trie.CommitTo(batch) + root, err := b.trie.Commit(nil) if err != nil { return err - } else { - batch.Write() - sectionHead := b.sectionHeads[b.bloomTrieRatio-1] - log.Info("Storing BloomTrie", "section", b.section, "sectionHead", fmt.Sprintf("%064x", sectionHead), "root", fmt.Sprintf("%064x", root), "compression ratio", float64(compSize)/float64(decompSize)) - StoreBloomTrieRoot(b.db, b.section, sectionHead, root) } + b.triedb.Commit(root, false) + + sectionHead := b.sectionHeads[b.bloomTrieRatio-1] + log.Info("Storing BloomTrie", "section", b.section, "sectionHead", fmt.Sprintf("%064x", sectionHead), "root", fmt.Sprintf("%064x", root), "compression ratio", float64(compSize)/float64(decompSize)) + StoreBloomTrieRoot(b.diskdb, b.section, sectionHead, root) return nil } diff --git a/light/trie.go b/light/trie.go index 7a9c86b98..c07e99461 100644 --- a/light/trie.go +++ b/light/trie.go @@ -18,12 +18,14 @@ package light import ( "context" + "errors" "fmt" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/trie" ) @@ -83,6 +85,10 @@ func (db *odrDatabase) ContractCodeSize(addrHash, codeHash common.Hash) (int, er return len(code), err } +func (db *odrDatabase) TrieDB() *trie.Database { + return nil +} + type odrTrie struct { db *odrDatabase id *TrieID @@ -113,11 +119,11 @@ func (t *odrTrie) TryDelete(key []byte) error { }) } -func (t *odrTrie) CommitTo(db trie.DatabaseWriter) (common.Hash, error) { +func (t *odrTrie) Commit(onleaf trie.LeafCallback) (common.Hash, error) { if t.trie == nil { return t.id.Root, nil } - return t.trie.CommitTo(db) + return t.trie.Commit(onleaf) } func (t *odrTrie) Hash() common.Hash { @@ -135,13 +141,17 @@ func (t *odrTrie) GetKey(sha []byte) []byte { return nil } +func (t *odrTrie) Prove(key []byte, fromLevel uint, proofDb ethdb.Putter) error { + return errors.New("not implemented, needs client/server interface split") +} + // do tries and retries to execute a function until it returns with no error or // an error type other than MissingNodeError 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, trie.NewDatabase(t.db.backend.Database())) } if err == nil { err = fn() @@ -167,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, t.db.backend.Database()) + t, err := trie.New(t.id.Root, trie.NewDatabase(t.db.backend.Database())) if err == nil { it.t.trie = t } diff --git a/light/trie_test.go b/light/trie_test.go index d99664718..0d6b2cc1d 100644 --- a/light/trie_test.go +++ b/light/trie_test.go @@ -40,7 +40,7 @@ func TestNodeIterator(t *testing.T) { genesis = gspec.MustCommit(fulldb) ) gspec.MustCommit(lightdb) - blockchain, _ := core.NewBlockChain(fulldb, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}) + blockchain, _ := core.NewBlockChain(fulldb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}) gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), fulldb, 4, testChainGen) if _, err := blockchain.InsertChain(gchain); err != nil { panic(err) diff --git a/light/txpool_test.go b/light/txpool_test.go index b343f79b0..13d7d3ceb 100644 --- a/light/txpool_test.go +++ b/light/txpool_test.go @@ -88,7 +88,7 @@ func TestTxPool(t *testing.T) { ) gspec.MustCommit(ldb) // Assemble the test environment - blockchain, _ := core.NewBlockChain(sdb, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}) + blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}) gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), sdb, poolTestBlocks, txPoolTestChainGen) if _, err := blockchain.InsertChain(gchain); err != nil { panic(err) -- cgit v1.2.3