diff --git a/core/blockchain.go b/core/blockchain.go index 794e1915fc..ecae115d7b 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -104,6 +104,8 @@ type BlockChain struct { procInterrupt int32 // interrupt signaler for block processing wg sync.WaitGroup // chain processing wait group for shutting down + chainProcessors []ChainProcessor // NewHead function called under chain lock after a new head has been written to the db + engine consensus.Engine processor Processor // block processor interface validator Validator // block and state validator interface @@ -168,6 +170,15 @@ func NewBlockChain(chainDb ethdb.Database, config *params.ChainConfig, engine co return bc, nil } +// AddChainProcessor adds a new head callback function that will be +// called under chain lock after a new head has been written to the db +func (self *BlockChain) AddChainProcessor(cp ChainProcessor) { + self.chainmu.Lock() + self.chainProcessors = append(self.chainProcessors, cp) + cp.NewHead(self.hc.CurrentHeader().Number.Uint64(), false) + self.chainmu.Unlock() +} + func (self *BlockChain) getProcInterrupt() bool { return atomic.LoadInt32(&self.procInterrupt) == 1 } @@ -243,6 +254,7 @@ func (bc *BlockChain) SetHead(head uint64) error { bc.mu.Lock() defer bc.mu.Unlock() + oldHead := bc.hc.CurrentHeader().Number.Uint64() // Rewind the header chain, deleting all block bodies until then delFn := func(hash common.Hash, num uint64) { DeleteBody(bc.chainDb, hash, num) @@ -256,6 +268,11 @@ func (bc *BlockChain) SetHead(head uint64) error { bc.blockCache.Purge() bc.futureBlocks.Purge() + if head < oldHead { + for _, cp := range bc.chainProcessors { + cp.NewHead(head, true) + } + } // Rewind the block chain, ensuring we don't end up with a stateless head block if bc.currentBlock != nil && currentHeader.Number.Uint64() < bc.currentBlock.NumberU64() { bc.currentBlock = bc.GetBlock(currentHeader.Hash(), currentHeader.Number.Uint64()) @@ -637,12 +654,15 @@ func (self *BlockChain) Rollback(chain []common.Hash) { self.mu.Lock() defer self.mu.Unlock() + var rollbackHead *types.Header + for i := len(chain) - 1; i >= 0; i-- { hash := chain[i] currentHeader := self.hc.CurrentHeader() if currentHeader.Hash() == hash { - self.hc.SetCurrentHeader(self.GetHeader(currentHeader.ParentHash, currentHeader.Number.Uint64()-1)) + rollbackHead = self.GetHeader(currentHeader.ParentHash, currentHeader.Number.Uint64()-1) + self.hc.SetCurrentHeader(rollbackHead) } if self.currentFastBlock.Hash() == hash { self.currentFastBlock = self.GetBlock(self.currentFastBlock.ParentHash(), self.currentFastBlock.NumberU64()-1) @@ -653,6 +673,12 @@ func (self *BlockChain) Rollback(chain []common.Hash) { WriteHeadBlockHash(self.chainDb, self.currentBlock.Hash()) } } + + if rollbackHead != nil { + for _, cp := range self.chainProcessors { + cp.NewHead(rollbackHead.Number.Uint64(), true) + } + } } // SetReceiptsData computes all the non-consensus fields of the receipts @@ -1316,7 +1342,16 @@ func (self *BlockChain) InsertHeaderChain(chain []*types.Header, checkFreq int) self.mu.Lock() defer self.mu.Unlock() - _, err := self.hc.WriteHeader(header) + _, err := self.hc.WriteHeader(header, func(head *types.Header) { + for _, cp := range self.chainProcessors { + cp.NewHead(head.Number.Uint64(), true) + } + }) + + for _, cp := range self.chainProcessors { + cp.NewHead(header.Number.Uint64(), false) + } + return err } @@ -1339,7 +1374,7 @@ func (self *BlockChain) writeHeader(header *types.Header) error { self.mu.Lock() defer self.mu.Unlock() - _, err := self.hc.WriteHeader(header) + _, err := self.hc.WriteHeader(header, nil) return err } diff --git a/core/chain_processor.go b/core/chain_processor.go new file mode 100644 index 0000000000..d431217137 --- /dev/null +++ b/core/chain_processor.go @@ -0,0 +1,226 @@ +// 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 core implements the Ethereum consensus protocol. +package core + +import ( + "encoding/binary" + "sync" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethdb" +) + +// ChainProcessor is an external process that creates auxiliary data structures +// using the canonical chain as an input. Its callback function NewHead is called every +// time a new head is added to the chain or it gets rolled back. The callback function +// should never block because it is called under the chain's mutex lock so that the +// structures can stay consistent with the canonical chain. It should also not call +// any chain functions that use this mutex because that would cause a deadlock. +// The same interface can be used for chaining processors (one using the output of the +// other). +type ChainProcessor interface { + NewHead(headNum uint64, rollBack bool) +} + +// ChainIndexer is a ChainProcessor that does a post-processing job for +// equally sized sections of the canonical chain (like BlooomBits and CHT structures). +// Further child ChainProcessors can be added which use the output of this section +// processor. These child processors receive NewHead calls only after an entire section +// has been finished or in case of rollbacks that might affect already finished sections. +type ChainIndexer struct { + db ethdb.Database + validSectionsKey []byte + backend ChainIndex + sectionSize, confirmReq uint64 + stop chan struct{} + updateCh chan uint64 + lock sync.Mutex + calcValid bool + procWait time.Duration + stored, calcIdx, lastForwarded uint64 + childProcessors []ChainProcessor +} + +type ChainIndex interface { + Reset(section uint64) + Process(header *types.Header) + Commit(db ethdb.Database) error + UpdateMsg(done, all uint64) // print a progress update message if necessary (only called when multiple sections need to be processed) +} + +// NewChainIndexer creates a new ChainIndexer +// backend: an implementation of ChainIndex +// sectionSize: the size of processable sections +// confirmReq: required number of confirmation blocks before a new section is being processed +// procWait: waiting time between processing sections (simple way of limiting the resource usage of a db upgrade) +// stop: quit channel +func NewChainIndexer(db ethdb.Database, validSectionsKey []byte, backend ChainIndex, sectionSize, confirmReq uint64, procWait time.Duration, stop chan struct{}) *ChainIndexer { + c := &ChainIndexer{ + db: db, + validSectionsKey: validSectionsKey, + backend: backend, + sectionSize: sectionSize, + confirmReq: confirmReq, + stop: stop, + procWait: procWait, + updateCh: make(chan uint64, 100), + } + c.stored = c.ValidSections() + go c.updateLoop() + return c +} + +func (c *ChainIndexer) updateLoop() { + tryUpdate := make(chan struct{}, 1) + updating := false + var targetCount uint64 + updateMsg := false + + for { + select { + case <-c.stop: + return + case targetCount = <-c.updateCh: + if !updating { + updating = true + tryUpdate <- struct{}{} + } + case <-tryUpdate: + c.lock.Lock() + if targetCount > c.stored { + if !updateMsg && targetCount > c.stored+1 { + updateMsg = true + c.backend.UpdateMsg(c.stored, targetCount) + } + c.calcValid = true + c.calcIdx = c.stored + + c.lock.Unlock() + ok := c.processSection(c.calcIdx) + c.lock.Lock() + + if ok && c.calcValid { + c.stored = c.calcIdx + 1 + c.setValidSections(c.stored) + if updateMsg { + c.backend.UpdateMsg(c.stored, targetCount) + if c.stored >= targetCount { + updateMsg = false + } + } + c.lastForwarded = c.stored*c.sectionSize - 1 + for _, cp := range c.childProcessors { + cp.NewHead(c.lastForwarded, false) + } + } + c.calcValid = false + } + stored := c.stored + c.lock.Unlock() + + if targetCount > stored { + go func() { + time.Sleep(c.procWait) + tryUpdate <- struct{}{} + }() + } else { + updating = false + } + } + } +} + +func (c *ChainIndexer) processSection(section uint64) bool { + c.backend.Reset(section) + for i := section * c.sectionSize; i < (section+1)*c.sectionSize; i++ { + hash := GetCanonicalHash(c.db, i) + if hash == (common.Hash{}) { + return false + } + header := GetHeader(c.db, hash, i) + if header == nil { + return false + } + c.backend.Process(header) + } + return c.backend.Commit(c.db) == nil +} + +func (c *ChainIndexer) ValidSections() uint64 { + data, _ := c.db.Get(c.validSectionsKey) + if len(data) == 8 { + return binary.BigEndian.Uint64(data[:]) + } + return 0 +} + +func (c *ChainIndexer) setValidSections(cnt uint64) { + var data [8]byte + binary.BigEndian.PutUint64(data[:], cnt) + c.db.Put(c.validSectionsKey, data[:]) +} + +// NewHead implements the ChainProcessor interface +func (c *ChainIndexer) NewHead(headNum uint64, rollback bool) { + c.lock.Lock() + defer c.lock.Unlock() + + if rollback { + firstChanged := headNum / c.sectionSize + if firstChanged <= c.calcIdx { + c.calcValid = false + } + if firstChanged < c.stored { + c.stored = firstChanged + c.setValidSections(c.stored) + select { + case <-c.stop: + case c.updateCh <- firstChanged: + } + } + + if headNum < c.lastForwarded { + c.lastForwarded = headNum + for _, cp := range c.childProcessors { + cp.NewHead(c.lastForwarded, true) + } + } + + } else { + var newCount uint64 + if headNum >= c.confirmReq { + newCount = (headNum + 1 - c.confirmReq) / c.sectionSize + if newCount > c.stored { + go func() { + select { + case <-c.stop: + case c.updateCh <- newCount: + } + }() + } + } + } +} + +// AddChildProcessor adds a child ChainProcessor that can use the output of this +// section processor. +func (c *ChainIndexer) AddChildProcessor(cp ChainProcessor) { + c.childProcessors = append(c.childProcessors, cp) +} diff --git a/core/headerchain.go b/core/headerchain.go index 9bb7f17937..4dce739f3c 100644 --- a/core/headerchain.go +++ b/core/headerchain.go @@ -119,6 +119,10 @@ func (hc *HeaderChain) GetBlockNumber(hash common.Hash) uint64 { return number } +// ReorgCallback is a callback function to be called in case of a chain reorg. +// The parameter is the header of the last unchanged canonical block. +type ReorgCallback func(*types.Header) + // WriteHeader writes a header into the local chain, given that its parent is // already known. If the total difficulty of the newly inserted header becomes // greater than the current known TD, the canonical chain is re-routed. @@ -128,7 +132,7 @@ func (hc *HeaderChain) GetBlockNumber(hash common.Hash) uint64 { // without the real blocks. Hence, writing headers directly should only be done // in two scenarios: pure-header mode of operation (light clients), or properly // separated header/block phases (non-archive clients). -func (hc *HeaderChain) WriteHeader(header *types.Header) (status WriteStatus, err error) { +func (hc *HeaderChain) WriteHeader(header *types.Header, reorgCallback ReorgCallback) (status WriteStatus, err error) { // Cache some values to prevent constant recalculation var ( hash = header.Hash() @@ -174,6 +178,11 @@ func (hc *HeaderChain) WriteHeader(header *types.Header) (status WriteStatus, er headNumber = headHeader.Number.Uint64() - 1 headHeader = hc.GetHeader(headHash, headNumber) } + + if reorgCallback != nil && headNumber < hc.currentHeader.Number.Uint64() { + reorgCallback(headHeader) + } + // Extend the canonical chain with the new header if err := WriteCanonicalHash(hc.chainDb, hash, number); err != nil { log.Crit("Failed to insert header number", "err", err) diff --git a/light/lightchain.go b/light/lightchain.go index 5b7e570417..26517b9b40 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -64,6 +64,8 @@ type LightChain struct { procInterrupt int32 // interrupt signaler for block processing wg sync.WaitGroup + chainProcessors []core.ChainProcessor // NewHead function called under chain lock after a new head has been written to the db + engine consensus.Engine } @@ -114,6 +116,15 @@ func NewLightChain(odr OdrBackend, config *params.ChainConfig, engine consensus. return bc, nil } +// AddChainProcessor adds a new head callback function that will be +// called under chain lock after a new head has been written to the db +func (self *LightChain) AddChainProcessor(cp core.ChainProcessor) { + self.chainmu.Lock() + self.chainProcessors = append(self.chainProcessors, cp) + cp.NewHead(self.hc.CurrentHeader().Number.Uint64(), false) + self.chainmu.Unlock() +} + func (self *LightChain) getProcInterrupt() bool { return atomic.LoadInt32(&self.procInterrupt) == 1 } @@ -149,7 +160,13 @@ func (bc *LightChain) SetHead(head uint64) { bc.mu.Lock() defer bc.mu.Unlock() + oldHead := bc.hc.CurrentHeader().Number.Uint64() bc.hc.SetHead(head, nil) + if head < oldHead { + for _, cp := range bc.chainProcessors { + cp.NewHead(head, true) + } + } bc.loadLastState() } @@ -312,11 +329,20 @@ func (self *LightChain) Rollback(chain []common.Hash) { self.mu.Lock() defer self.mu.Unlock() + var rollbackHead *types.Header + for i := len(chain) - 1; i >= 0; i-- { hash := chain[i] if head := self.hc.CurrentHeader(); head.Hash() == hash { - self.hc.SetCurrentHeader(self.GetHeader(head.ParentHash, head.Number.Uint64()-1)) + rollbackHead = self.GetHeader(head.ParentHash, head.Number.Uint64()-1) + self.hc.SetCurrentHeader(rollbackHead) + } + } + + if rollbackHead != nil { + for _, cp := range self.chainProcessors { + cp.NewHead(rollbackHead.Number.Uint64(), true) } } } @@ -367,7 +393,15 @@ func (self *LightChain) InsertHeaderChain(chain []*types.Header, checkFreq int) self.mu.Lock() defer self.mu.Unlock() - status, err := self.hc.WriteHeader(header) + status, err := self.hc.WriteHeader(header, func(head *types.Header) { + for _, cp := range self.chainProcessors { + cp.NewHead(head.Number.Uint64(), true) + } + }) + + for _, cp := range self.chainProcessors { + cp.NewHead(header.Number.Uint64(), false) + } switch status { case core.CanonStatTy: