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.
This commit is contained in:
Felix Lange 2017-09-07 19:35:27 +02:00
parent 1436f5111e
commit d61e09433a
2 changed files with 37 additions and 43 deletions

View file

@ -784,7 +784,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
} }
// WriteBlock writes the block to the chain. // 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) bc.wg.Add(1)
defer bc.wg.Done() 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 { if err := bc.hc.WriteTd(block.Hash(), block.NumberU64(), externTd); err != nil {
log.Crit("Failed to write block total difficulty", "err", err) 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) 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 // 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. // 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 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 status = CanonStatTy
} else { } else {
status = SideStatTy 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()) bc.futureBlocks.Remove(block.Hash())
return status, nil
return
} }
// InsertChain will attempt to insert the given chain in to the canonical chain or, otherwise, create a fork. If an error is returned // 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) bc.reportBlock(block, receipts, err)
return i, 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 // Write the block to the chain and get the status.
coalescedLogs = append(coalescedLogs, logs...) status, err := bc.WriteBlockAndState(block, receipts, state)
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)
if err != nil { if err != nil {
return i, err return i, err
} }
switch status { switch status {
case CanonStatTy: case CanonStatTy:
log.Debug("Inserted new block", "number", block.Number(), "hash", block.Hash(), "uncles", len(block.Uncles()), 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))) "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) blockInsertTimer.UpdateSince(bstart)
events = append(events, ChainEvent{block, block.Hash(), logs}) events = append(events, ChainEvent{block, block.Hash(), logs})
// We need some control over the mining operation. Acquiring locks and waiting // 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() { if bc.LastBlockHash() == block.Hash() {
events = append(events, ChainHeadEvent{block}) 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: case SideStatTy:
log.Debug("Inserted forked block", "number", block.Number(), "hash", block.Hash(), "diff", block.Difficulty(), "elapsed", 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())) common.PrettyDuration(time.Since(bstart)), "txs", len(block.Transactions()), "gas", block.GasUsed(), "uncles", len(block.Uncles()))

View file

@ -304,13 +304,8 @@ func (self *worker) wait() {
} }
go self.mux.Post(core.NewMinedBlockEvent{Block: block}) go self.mux.Post(core.NewMinedBlockEvent{Block: block})
} else { } else {
work.state.CommitTo(self.chainDb, self.config.IsEIP158(block.Number())) // Update the block hash in all logs since it is now available and not when the
stat, err := self.chain.WriteBlock(block) // receipt/log of individual transactions were created.
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
for _, r := range work.receipts { for _, r := range work.receipts {
for _, l := range r.Logs { for _, l := range r.Logs {
l.BlockHash = block.Hash() l.BlockHash = block.Hash()
@ -319,15 +314,17 @@ func (self *worker) wait() {
for _, log := range work.state.Logs() { for _, log := range work.state.Logs() {
log.BlockHash = block.Hash() 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 // check if canon block and write transactions
if stat == core.CanonStatTy { if stat == core.CanonStatTy {
// This puts transactions in a extra db for rpc
core.WriteTxLookupEntries(self.chainDb, block)
// implicit by posting ChainHeadEvent // implicit by posting ChainHeadEvent
mustCommitNewWork = false mustCommitNewWork = false
} }
// broadcast before waiting for validation // broadcast before waiting for validation
go func(block *types.Block, logs []*types.Log, receipts []*types.Receipt) { go func(block *types.Block, logs []*types.Log, receipts []*types.Receipt) {
self.mux.Post(core.NewMinedBlockEvent{Block: block}) self.mux.Post(core.NewMinedBlockEvent{Block: block})
@ -336,16 +333,12 @@ func (self *worker) wait() {
coalescedLogs []*types.Log coalescedLogs []*types.Log
) )
events = append(events, core.ChainEvent{Block: block, Hash: block.Hash(), Logs: logs}) events = append(events, core.ChainEvent{Block: block, Hash: block.Hash(), Logs: logs})
if stat == core.CanonStatTy { if stat == core.CanonStatTy {
events = append(events, core.ChainHeadEvent{Block: block}) events = append(events, core.ChainHeadEvent{Block: block})
coalescedLogs = logs coalescedLogs = logs
} }
// post blockchain events // post blockchain events
self.chain.PostChainEvents(events, coalescedLogs) 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) }(block, work.state.Logs(), work.receipts)
} }
// Insert the block into the set of pending ones to wait for confirmations // Insert the block into the set of pending ones to wait for confirmations