core: parallelise nonce checking when processing blocks

ChainManager now uses a parallel approach to block processing where all
nonces are checked seperatly from the block processing process. This
speeds up the process by about 3 times on my i7
This commit is contained in:
obscuren 2015-05-17 01:42:30 +02:00
parent 443d024843
commit c67424ecc8
4 changed files with 94 additions and 15 deletions

View File

@ -336,8 +336,8 @@ func GetChain(ctx *cli.Context) (*core.ChainManager, common.Database, common.Dat
} }
eventMux := new(event.TypeMux) eventMux := new(event.TypeMux)
chainManager := core.NewChainManager(blockDb, stateDb, eventMux)
pow := ethash.New() pow := ethash.New()
chainManager := core.NewChainManager(blockDb, stateDb, pow, eventMux)
txPool := core.NewTxPool(eventMux, chainManager.State, chainManager.GasLimit) txPool := core.NewTxPool(eventMux, chainManager.State, chainManager.GasLimit)
blockProcessor := core.NewBlockProcessor(stateDb, extraDb, pow, txPool, chainManager, eventMux) blockProcessor := core.NewBlockProcessor(stateDb, extraDb, pow, txPool, chainManager, eventMux)
chainManager.SetProcessor(blockProcessor) chainManager.SetProcessor(blockProcessor)

View File

@ -189,7 +189,7 @@ func (sm *BlockProcessor) processWithParent(block, parent *types.Block) (logs st
state := state.New(parent.Root(), sm.db) state := state.New(parent.Root(), sm.db)
// Block validation // Block validation
if err = sm.ValidateHeader(block.Header(), parent.Header()); err != nil { if err = sm.ValidateHeader(block.Header(), parent.Header(), false); err != nil {
return return
} }
@ -269,7 +269,7 @@ func (sm *BlockProcessor) processWithParent(block, parent *types.Block) (logs st
// Validates the current block. Returns an error if the block was invalid, // Validates the current block. Returns an error if the block was invalid,
// an uncle or anything that isn't on the current block chain. // an uncle or anything that isn't on the current block chain.
// Validation validates easy over difficult (dagger takes longer time = difficult) // Validation validates easy over difficult (dagger takes longer time = difficult)
func (sm *BlockProcessor) ValidateHeader(block, parent *types.Header) error { func (sm *BlockProcessor) ValidateHeader(block, parent *types.Header, checkPow bool) error {
if big.NewInt(int64(len(block.Extra))).Cmp(params.MaximumExtraDataSize) == 1 { if big.NewInt(int64(len(block.Extra))).Cmp(params.MaximumExtraDataSize) == 1 {
return fmt.Errorf("Block extra data too long (%d)", len(block.Extra)) return fmt.Errorf("Block extra data too long (%d)", len(block.Extra))
} }
@ -300,9 +300,11 @@ func (sm *BlockProcessor) ValidateHeader(block, parent *types.Header) error {
return BlockEqualTSErr //ValidationError("Block timestamp equal or less than previous block (%v - %v)", block.Time, parent.Time) return BlockEqualTSErr //ValidationError("Block timestamp equal or less than previous block (%v - %v)", block.Time, parent.Time)
} }
// Verify the nonce of the block. Return an error if it's not valid if checkPow {
if !sm.Pow.Verify(types.NewBlockWithHeader(block)) { // Verify the nonce of the block. Return an error if it's not valid
return ValidationError("Block's nonce is invalid (= %x)", block.Nonce) if !sm.Pow.Verify(types.NewBlockWithHeader(block)) {
return ValidationError("Block's nonce is invalid (= %x)", block.Nonce)
}
} }
return nil return nil
@ -358,7 +360,7 @@ func (sm *BlockProcessor) VerifyUncles(statedb *state.StateDB, block, parent *ty
return UncleError("uncle[%d](%x)'s parent unknown (%x)", i, hash[:4], uncle.ParentHash[0:4]) return UncleError("uncle[%d](%x)'s parent unknown (%x)", i, hash[:4], uncle.ParentHash[0:4])
} }
if err := sm.ValidateHeader(uncle, ancestorHeaders[uncle.ParentHash]); err != nil { if err := sm.ValidateHeader(uncle, ancestorHeaders[uncle.ParentHash], true); err != nil {
return ValidationError(fmt.Sprintf("uncle[%d](%x) header invalid: %v", i, hash[:4], err)) return ValidationError(fmt.Sprintf("uncle[%d](%x) header invalid: %v", i, hash[:4], err))
} }
} }

View File

@ -5,6 +5,7 @@ import (
"fmt" "fmt"
"io" "io"
"math/big" "math/big"
"runtime"
"sync" "sync"
"time" "time"
@ -15,6 +16,7 @@ import (
"github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/logger/glog"
"github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/pow"
"github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rlp"
) )
@ -100,9 +102,11 @@ type ChainManager struct {
quit chan struct{} quit chan struct{}
wg sync.WaitGroup wg sync.WaitGroup
pow pow.PoW
} }
func NewChainManager(blockDb, stateDb common.Database, mux *event.TypeMux) *ChainManager { func NewChainManager(blockDb, stateDb common.Database, pow pow.PoW, mux *event.TypeMux) *ChainManager {
bc := &ChainManager{ bc := &ChainManager{
blockDb: blockDb, blockDb: blockDb,
stateDb: stateDb, stateDb: stateDb,
@ -110,6 +114,7 @@ func NewChainManager(blockDb, stateDb common.Database, mux *event.TypeMux) *Chai
eventMux: mux, eventMux: mux,
quit: make(chan struct{}), quit: make(chan struct{}),
cache: NewBlockCache(blockCacheLimit), cache: NewBlockCache(blockCacheLimit),
pow: pow,
} }
bc.setLastState() bc.setLastState()
@ -529,10 +534,19 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
stats struct{ queued, processed, ignored int } stats struct{ queued, processed, ignored int }
tstart = time.Now() tstart = time.Now()
) )
// check the nonce in parallel to the block processing
// this speeds catching up significantly
nonceErrCh := make(chan error)
go func() {
nonceErrCh <- verifyNonces(self.pow, chain)
}()
for i, block := range chain { for i, block := range chain {
if block == nil { if block == nil {
continue continue
} }
// Setting block.Td regardless of error (known for example) prevents errors down the line // Setting block.Td regardless of error (known for example) prevents errors down the line
// in the protocol handler // in the protocol handler
block.Td = new(big.Int).Set(CalcTD(block, self.GetBlock(block.ParentHash()))) block.Td = new(big.Int).Set(CalcTD(block, self.GetBlock(block.ParentHash())))
@ -562,11 +576,7 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
continue continue
} }
h := block.Header() blockErr(block, err)
glog.V(logger.Error).Infof("INVALID block #%v (%x)\n", h.Number, h.Hash().Bytes())
glog.V(logger.Error).Infoln(err)
glog.V(logger.Debug).Infoln(block)
return i, err return i, err
} }
@ -620,6 +630,13 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
} }
// check and wait for the nonce error channel and
// make sure no nonce error was thrown in the process
err := <-nonceErrCh
if err != nil {
return 0, err
}
if (stats.queued > 0 || stats.processed > 0 || stats.ignored > 0) && bool(glog.V(logger.Info)) { if (stats.queued > 0 || stats.processed > 0 || stats.ignored > 0) && bool(glog.V(logger.Info)) {
tend := time.Since(tstart) tend := time.Since(tstart)
start, end := chain[0], chain[len(chain)-1] start, end := chain[0], chain[len(chain)-1]
@ -718,3 +735,63 @@ out:
} }
} }
} }
func blockErr(block *types.Block, err error) {
h := block.Header()
glog.V(logger.Error).Infof("INVALID block #%v (%x)\n", h.Number, h.Hash().Bytes())
glog.V(logger.Error).Infoln(err)
glog.V(logger.Debug).Infoln(block)
}
// verifyNonces verifies nonces of the given blocks in parallel and returns
// an error if one of the blocks nonce verifications failed.
func verifyNonces(pow pow.PoW, blocks []*types.Block) error {
// Spawn a few workers. They listen for blocks on the in channel
// and send results on done. The workers will exit in the
// background when in is closed.
var (
in = make(chan *types.Block)
done = make(chan error, runtime.GOMAXPROCS(0))
)
defer close(in)
for i := 0; i < cap(done); i++ {
go verifyNonce(pow, in, done)
}
// Feed blocks to the workers, aborting at the first invalid nonce.
var (
running, i int
block *types.Block
sendin = in
)
for i < len(blocks) || running > 0 {
if i == len(blocks) {
// Disable sending to in.
sendin = nil
} else {
block = blocks[i]
i++
}
select {
case sendin <- block:
running++
case err := <-done:
running--
if err != nil {
return err
}
}
}
return nil
}
// verifyNonce is a worker for the verifyNonces method. It will run until
// in is closed.
func verifyNonce(pow pow.PoW, in <-chan *types.Block, done chan<- error) {
for block := range in {
if !pow.Verify(block) {
done <- ValidationError("Block's nonce is invalid (= %x)", block.Nonce)
} else {
done <- nil
}
}
}

View File

@ -266,9 +266,9 @@ func New(config *Config) (*Ethereum, error) {
MinerThreads: config.MinerThreads, MinerThreads: config.MinerThreads,
} }
eth.chainManager = core.NewChainManager(blockDb, stateDb, eth.EventMux())
eth.downloader = downloader.New(eth.EventMux(), eth.chainManager.HasBlock, eth.chainManager.GetBlock)
eth.pow = ethash.New() eth.pow = ethash.New()
eth.chainManager = core.NewChainManager(blockDb, stateDb, eth.pow, eth.EventMux())
eth.downloader = downloader.New(eth.EventMux(), eth.chainManager.HasBlock, eth.chainManager.GetBlock)
eth.txPool = core.NewTxPool(eth.EventMux(), eth.chainManager.State, eth.chainManager.GasLimit) eth.txPool = core.NewTxPool(eth.EventMux(), eth.chainManager.State, eth.chainManager.GasLimit)
eth.blockProcessor = core.NewBlockProcessor(stateDb, extraDb, eth.pow, eth.txPool, eth.chainManager, eth.EventMux()) eth.blockProcessor = core.NewBlockProcessor(stateDb, extraDb, eth.pow, eth.txPool, eth.chainManager, eth.EventMux())
eth.chainManager.SetProcessor(eth.blockProcessor) eth.chainManager.SetProcessor(eth.blockProcessor)