Merge pull request #1151 from fjl/parallel-nonce-2

core: re-add parallel nonce checks

1 files changed, 49 insertions, 67 deletions

diff --git a/core/chain_manager.go b/core/chain_manager.go
index 088ca8d5b..927055103 100644
--- a/core/chain_manager.go
+++ b/core/chain_manager.go
@@ -522,13 +522,14 @@ type queueEvent struct {
 }
 
 func (self *ChainManager) procFutureBlocks() {
-	blocks := []*types.Block{}
+	var blocks []*types.Block
 	self.futureBlocks.Each(func(i int, block *types.Block) {
 		blocks = append(blocks, block)
 	})
-
-	types.BlockBy(types.Number).Sort(blocks)
-	self.InsertChain(blocks)
+	if len(blocks) > 0 {
+		types.BlockBy(types.Number).Sort(blocks)
+		self.InsertChain(blocks)
+	}
 }
 
 // InsertChain will attempt to insert the given chain in to the canonical chain or, otherwise, create a fork. It an error is returned
@@ -540,17 +541,34 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
 	self.chainmu.Lock()
 	defer self.chainmu.Unlock()
 
-	// A queued approach to delivering events. This is generally faster than direct delivery and requires much less mutex acquiring.
+	// A queued approach to delivering events. This is generally
+	// faster than direct delivery and requires much less mutex
+	// acquiring.
 	var (
 		queue      = make([]interface{}, len(chain))
 		queueEvent = queueEvent{queue: queue}
 		stats      struct{ queued, processed, ignored int }
 		tstart     = time.Now()
+
+		nonceDone    = make(chan nonceResult, len(chain))
+		nonceQuit    = make(chan struct{})
+		nonceChecked = make([]bool, len(chain))
 	)
 
+	// Start the parallel nonce verifier.
+	go verifyNonces(self.pow, chain, nonceQuit, nonceDone)
+	defer close(nonceQuit)
+
 	for i, block := range chain {
-		if block == nil {
-			continue
+		// Wait for block i's nonce to be verified before processing
+		// its state transition.
+		for nonceChecked[i] {
+			r := <-nonceDone
+			nonceChecked[r.i] = true
+			if !r.valid {
+				block := chain[i]
+				return i, ValidationError("Block (#%v / %x) nonce is invalid (= %x)", block.Number(), block.Hash(), block.Nonce)
+			}
 		}
 
 		if BadHashes[block.Hash()] {
@@ -559,10 +577,6 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
 			return i, err
 		}
 
-		// create a nonce channel for parallisation of the nonce check
-		nonceErrCh := make(chan error)
-		go verifyBlockNonce(self.pow, block, nonceErrCh)
-
 		// Setting block.Td regardless of error (known for example) prevents errors down the line
 		// in the protocol handler
 		block.Td = new(big.Int).Set(CalcTD(block, self.GetBlock(block.ParentHash())))
@@ -571,9 +585,6 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
 		// all others will fail too (unless a known block is returned).
 		logs, err := self.processor.Process(block)
 		if err != nil {
-			// empty the nonce channel
-			<-nonceErrCh
-
 			if IsKnownBlockErr(err) {
 				stats.ignored++
 				continue
@@ -597,11 +608,6 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
 
 			return i, err
 		}
-		// Wait and check nonce channel and make sure it checks out fine
-		// otherwise return the error
-		if err := <-nonceErrCh; err != nil {
-			return i, err
-		}
 
 		cblock := self.currentBlock
 		// Compare the TD of the last known block in the canonical chain to make sure it's greater.
@@ -776,66 +782,42 @@ func blockErr(block *types.Block, err error) {
 	h := block.Header()
 	glog.V(logger.Error).Infof("Bad block #%v (%x)\n", h.Number, h.Hash().Bytes())
 	glog.V(logger.Error).Infoln(err)
-	glog.V(logger.Debug).Infoln(block)
+	glog.V(logger.Debug).Infoln(verifyNonces)
 }
 
-// verifyNonces verifies nonces of the given blocks in parallel and returns
+type nonceResult struct {
+	i     int
+	valid bool
+}
+
+// block 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 {
+func verifyNonces(pow pow.PoW, blocks []*types.Block, quit <-chan struct{}, done chan<- nonceResult) {
 	// 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))
+		in       = make(chan int)
+		nworkers = runtime.GOMAXPROCS(0)
 	)
 	defer close(in)
-	for i := 0; i < cap(done); i++ {
-		go verifyNonce(pow, in, done)
+	if len(blocks) < nworkers {
+		nworkers = len(blocks)
 	}
-	// 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
+	for i := 0; i < nworkers; i++ {
+		go func() {
+			for i := range in {
+				done <- nonceResult{i: i, valid: pow.Verify(blocks[i])}
 			}
-		}
+		}()
 	}
-	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 (#%v / %x) nonce is invalid (= %x)", block.Number(), block.Hash(), block.Nonce)
-		} else {
-			done <- nil
+	// Feed block indices to the workers.
+	for i := range blocks {
+		select {
+		case in <- i:
+			continue
+		case <-quit:
+			return
 		}
 	}
 }
-
-func verifyBlockNonce(pow pow.PoW, block *types.Block, done chan<- error) {
-	if !pow.Verify(block) {
-		done <- ValidationError("Block (#%v / %x) nonce is invalid (= %x)", block.Number(), block.Hash(), block.Nonce)
-	} else {
-		done <- nil
-	}
-}