Merge 3d8cb46e04 into 07aae19e5d

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
 }
 

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 <http://www.gnu.org/licenses/>.
+
+// 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)
+}

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)

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: