core: added nonce workers for load balancer

core/block_work.go

@@ -0,0 +1,51 @@
+package core
+
+import (
+	"math"
+
+	"github.com/ethereum/go-ethereum/balancer"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/pow"
+)
+
+type nonceResult struct {
+	index int
+	valid bool
+}
+
+func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.PoW) (chan nonceResult, chan struct{}) {
+	const workSize = 64
+
+	var (
+		nonceResults = make(chan nonceResult, len(blocks))                                      // the nonce result channel (buffered)
+		errch        = make(chan error, int(math.Ceil(float64(len(blocks))/float64(workSize)))) // error channel (buffered)
+	)
+	for i := 0; i < len(blocks); i += workSize {
+		max := int(math.Min(float64(i+workSize), float64(len(blocks)))) // get max size...
+		batch := blocks[i:max]                                          // ...and create batch
+
+		batchNo := i // batch number for task
+		// create new tasks
+		task := balancer.NewTask(func() error {
+			for i := 0; i < max-batchNo; i++ {
+				nonceResults <- nonceResult{batchNo + i, checker.Verify(batch[i])}
+			}
+			return nil
+		}, errch)
+
+		b.Push(task)
+	}
+
+	donech := make(chan struct{})
+	// we aren't at all interested in the errors
+	// since we handle errors ourself.
+	go func() {
+		<-donech // wait for parent proc to finish
+		for i := 0; i < cap(errch); i++ {
+			<-errch
+		}
+		close(errch)
+	}()
+
+	return nonceResults, donech
+}

core/blockchain.go

@@ -30,6 +30,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/ethereum/go-ethereum/balancer"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -113,6 +114,7 @@ type BlockChain struct {
 	rand      *mrand.Rand
 	processor Processor
 	validator Validator
+	balancer  *balancer.Balancer
 }
 
 // NewBlockChain returns a fully initialised block chain using information
@@ -137,6 +139,7 @@ func NewBlockChain(chainDb ethdb.Database, pow pow.PoW, mux *event.TypeMux) (*Bl
 		blockCache:   blockCache,
 		futureBlocks: futureBlocks,
 		pow:          pow,
+		balancer:     balancer.New(runtime.NumCPU()),
 	}
 	// Seed a fast but crypto originating random generator
 	seed, err := crand.Int(crand.Reader, big.NewInt(math.MaxInt64))
@@ -1122,8 +1125,10 @@ func (self *BlockChain) InsertChain(chain types.Blocks) (int, error) {
 	)
 
 	// Start the parallel nonce verifier.
-	nonceAbort, nonceResults := verifyNoncesFromBlocks(self.pow, chain)
+	//nonceAbort, nonceResults := verifyNoncesFromBlocks(self.pow, chain)
-	defer close(nonceAbort)
+	//defer close(nonceAbort)
+	nonceResults, donech := balanceBlockWork(self.balancer, chain, self.pow) // ...balance out work
+	defer close(donech)
 
 	txcount := 0
 	for i, block := range chain {
@@ -1252,6 +1257,8 @@ func (self *BlockChain) InsertChain(chain types.Blocks) (int, error) {
 		}
 		stats.processed++
 	}
+	// close the bogus errch. errors are delivered using owned channel
+	//close(errch) // we don't need to bother with checking
 
 	if (stats.queued > 0 || stats.processed > 0 || stats.ignored > 0) && bool(glog.V(logger.Info)) {
 		tend := time.Since(tstart)