From d61e09433a34fbd436c460181fb27436760ad526 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Thu, 7 Sep 2017 19:35:27 +0200 Subject: [PATCH] core: write imported block data using a batch Restore batch writes of state and add blocks, tx entries, receipts to the same batch. The change also simplifies the miner. This commit also removes posting of logs when a forked block is imported. --- core/blockchain.go | 59 +++++++++++++++++++++++----------------------- miner/worker.go | 21 ++++++----------- 2 files changed, 37 insertions(+), 43 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index fcbef17e53..bcf7c9af7d 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -784,7 +784,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ } // WriteBlock writes the block to the chain. -func (bc *BlockChain) WriteBlock(block *types.Block) (status WriteStatus, err error) { +func (bc *BlockChain) WriteBlockAndState(block *types.Block, receipts []*types.Receipt, state *state.StateDB) (status WriteStatus, err error) { bc.wg.Add(1) defer bc.wg.Done() @@ -809,9 +809,17 @@ func (bc *BlockChain) WriteBlock(block *types.Block) (status WriteStatus, err er if err := bc.hc.WriteTd(block.Hash(), block.NumberU64(), externTd); err != nil { log.Crit("Failed to write block total difficulty", "err", err) } - if err := WriteBlock(bc.chainDb, block); err != nil { + // Write other block data using a batch. + batch := bc.chainDb.NewBatch() + if err := WriteBlock(batch, block); err != nil { log.Crit("Failed to write block contents", "err", err) } + if _, err := state.CommitTo(batch, bc.config.IsEIP158(block.Number())); err != nil { + return NonStatTy, err + } + if err := WriteBlockReceipts(batch, block.Hash(), block.NumberU64(), receipts); err != nil { + return NonStatTy, err + } // If the total difficulty is higher than our known, add it to the canonical chain // Second clause in the if statement reduces the vulnerability to selfish mining. @@ -823,15 +831,28 @@ func (bc *BlockChain) WriteBlock(block *types.Block) (status WriteStatus, err er return NonStatTy, err } } - bc.insert(block) // Insert the block as the new head of the chain + // Write the positional metadata for transaction and receipt lookups + if err := WriteTxLookupEntries(batch, block); err != nil { + return NonStatTy, err + } + // Write hash preimages + if err := WritePreimages(bc.chainDb, block.NumberU64(), state.Preimages()); err != nil { + return NonStatTy, err + } status = CanonStatTy } else { status = SideStatTy } + if err := batch.Write(); err != nil { + return NonStatTy, err + } + // Set new head. + if status == CanonStatTy { + bc.insert(block) + } bc.futureBlocks.Remove(block.Hash()) - - return + return status, nil } // InsertChain will attempt to insert the given chain in to the canonical chain or, otherwise, create a fork. If an error is returned @@ -945,29 +966,18 @@ func (bc *BlockChain) InsertChain(chain types.Blocks) (int, error) { bc.reportBlock(block, receipts, err) return i, err } - // Write state changes to database - if _, err = state.CommitTo(bc.chainDb, bc.config.IsEIP158(block.Number())); err != nil { - return i, err - } - // coalesce logs for later processing - coalescedLogs = append(coalescedLogs, logs...) - - if err = WriteBlockReceipts(bc.chainDb, block.Hash(), block.NumberU64(), receipts); err != nil { - return i, err - } - - // write the block to the chain and get the status - status, err := bc.WriteBlock(block) + // Write the block to the chain and get the status. + status, err := bc.WriteBlockAndState(block, receipts, state) if err != nil { return i, err } - switch status { case CanonStatTy: log.Debug("Inserted new block", "number", block.Number(), "hash", block.Hash(), "uncles", len(block.Uncles()), "txs", len(block.Transactions()), "gas", block.GasUsed(), "elapsed", common.PrettyDuration(time.Since(bstart))) - + // coalesce logs for later processing + coalescedLogs = append(coalescedLogs, logs...) blockInsertTimer.UpdateSince(bstart) events = append(events, ChainEvent{block, block.Hash(), logs}) // We need some control over the mining operation. Acquiring locks and waiting @@ -976,15 +986,6 @@ func (bc *BlockChain) InsertChain(chain types.Blocks) (int, error) { if bc.LastBlockHash() == block.Hash() { events = append(events, ChainHeadEvent{block}) } - - // Write the positional metadata for transaction and receipt lookups - if err := WriteTxLookupEntries(bc.chainDb, block); err != nil { - return i, err - } - // Write hash preimages - if err := WritePreimages(bc.chainDb, block.NumberU64(), state.Preimages()); err != nil { - return i, err - } case SideStatTy: log.Debug("Inserted forked block", "number", block.Number(), "hash", block.Hash(), "diff", block.Difficulty(), "elapsed", common.PrettyDuration(time.Since(bstart)), "txs", len(block.Transactions()), "gas", block.GasUsed(), "uncles", len(block.Uncles())) diff --git a/miner/worker.go b/miner/worker.go index e1154ac068..43804d77bd 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -304,13 +304,8 @@ func (self *worker) wait() { } go self.mux.Post(core.NewMinedBlockEvent{Block: block}) } else { - work.state.CommitTo(self.chainDb, self.config.IsEIP158(block.Number())) - stat, err := self.chain.WriteBlock(block) - if err != nil { - log.Error("Failed writing block to chain", "err", err) - continue - } - // update block hash since it is now available and not when the receipt/log of individual transactions were created + // Update the block hash in all logs since it is now available and not when the + // receipt/log of individual transactions were created. for _, r := range work.receipts { for _, l := range r.Logs { l.BlockHash = block.Hash() @@ -319,15 +314,17 @@ func (self *worker) wait() { for _, log := range work.state.Logs() { log.BlockHash = block.Hash() } + stat, err := self.chain.WriteBlockAndState(block, work.receipts, work.state) + if err != nil { + log.Error("Failed writing block to chain", "err", err) + continue + } // check if canon block and write transactions if stat == core.CanonStatTy { - // This puts transactions in a extra db for rpc - core.WriteTxLookupEntries(self.chainDb, block) // implicit by posting ChainHeadEvent mustCommitNewWork = false } - // broadcast before waiting for validation go func(block *types.Block, logs []*types.Log, receipts []*types.Receipt) { self.mux.Post(core.NewMinedBlockEvent{Block: block}) @@ -336,16 +333,12 @@ func (self *worker) wait() { coalescedLogs []*types.Log ) events = append(events, core.ChainEvent{Block: block, Hash: block.Hash(), Logs: logs}) - if stat == core.CanonStatTy { events = append(events, core.ChainHeadEvent{Block: block}) coalescedLogs = logs } // post blockchain events self.chain.PostChainEvents(events, coalescedLogs) - if err := core.WriteBlockReceipts(self.chainDb, block.Hash(), block.NumberU64(), receipts); err != nil { - log.Warn("Failed writing block receipts", "err", err) - } }(block, work.state.Logs(), work.receipts) } // Insert the block into the set of pending ones to wait for confirmations