core: implemented parallel nonce checking & tx pub key derivation

The blockchain switched over to the generic tasks handler / load
balancer to check for block nonce, uncle nonce and transaction public
key gerivation.

balancer/balancer.go

@@ -2,7 +2,7 @@ package balancer
 
 import "container/heap"
 
-const maxTaskBuffer = 3000 // maximum amount of work a worker can have in its buffer
+var maxTaskBuffer = 3000 // maximum amount of work a worker can have in its buffer
 
 // Task repsents a single batch of work offered to a worker.
 type Task struct {
@@ -158,13 +158,14 @@ func (b *Balancer) dispatch(task Task) {
 	if len(w.tasks) == cap(w.tasks) {
 		heap.Push(&b.pool, w) // push full worker back to heap
 		// set the pending state (i.e. high load)
-		pending := (len(b.pool) - b.poolSize) * maxTaskBuffer
+		pending := (len(b.pool) - b.poolSize + 1) * maxTaskBuffer
 		// create the new temporary worker
 		worker := &Worker{
 			tasks:   make(chan Task, maxTaskBuffer),
 			temp:    true,
 			start:   pending,
 			pending: pending,
+			quit:    make(chan struct{}),
 		}
 		go worker.work(b.done) // spawn worker process
 		w = worker             // set new worker

balancer/balancer_test.go

@@ -1,6 +1,7 @@
 package balancer
 
 import (
+	"container/heap"
 	"math/big"
 	"testing"
 
@@ -9,6 +10,113 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 )
 
+func TestTemporaryWorker(t *testing.T) {
+	size := 5
+	maxTaskBuffer = 1
+	balancer := New(4)
+
+	qchan := make([]chan struct{}, size)
+	errch := make(chan error, size)
+
+	for i := 0; i < size; i++ {
+		i := i
+
+		qchan[i] = make(chan struct{})
+		balancer.dispatch(NewTask(func() error {
+			<-qchan[i] // blocks
+			return nil
+		}, errch))
+	}
+
+	if len(balancer.pool) != balancer.poolSize+1 {
+		t.Fatal("expected pool to be size", balancer.poolSize+1, "got", len(balancer.pool))
+	}
+
+	lastWorker := balancer.pool[len(balancer.pool)-1]
+	if !lastWorker.temp {
+		t.Error("expected last worker to be temp worker")
+	}
+
+	if lastWorker.start != (len(balancer.pool)-balancer.poolSize)*maxTaskBuffer {
+		t.Error("expected temp worker start", balancer.poolSize, "got", lastWorker.start)
+	}
+
+	// clean up
+	for i := 0; i < cap(qchan); i++ {
+		close(qchan[i])
+	}
+}
+
+func TestLoad(t *testing.T) {
+	maxTaskBuffer = 2
+	size := 5
+	balancer := New(4)
+
+	qchan := make([]chan struct{}, size)
+	errch := make(chan error, size)
+
+	for i := 0; i < size; i++ {
+		i := i
+
+		qchan[i] = make(chan struct{})
+		balancer.dispatch(NewTask(func() error {
+			<-qchan[i] // blocks
+			return nil
+		}, errch))
+	}
+
+	var foundTwo bool
+	for _, worker := range balancer.pool {
+		if worker.pending == 2 {
+			foundTwo = true
+		}
+	}
+	if !foundTwo {
+		t.Error("expected to have at least one item with pending 2")
+	}
+
+	// clean up
+	for i := 0; i < cap(qchan); i++ {
+		close(qchan[i])
+	}
+}
+
+func TestPool(t *testing.T) {
+	pool := make(Pool, 0)
+	heap.Init(&pool)
+
+	var workers = []*Worker{
+		&Worker{id: 0, pending: 0},
+		&Worker{id: 1, pending: 1},
+	}
+
+	for _, worker := range workers {
+		heap.Push(&pool, worker)
+	}
+
+	worker := heap.Pop(&pool).(*Worker)
+	if worker.id != 0 {
+		t.Error("expected worker 0 to be popped")
+	}
+
+	worker.pending = 2
+	heap.Push(&pool, worker)
+
+	if pool[worker.index] != worker {
+		t.Error("expected", worker.index, "to be at index but got different worker")
+	}
+
+	worker = heap.Pop(&pool).(*Worker)
+	if worker.id != 1 {
+		t.Error("expected worker 1 to be popped")
+	}
+	heap.Push(&pool, worker)
+
+	if pool[worker.index] != worker {
+		t.Error("expected", worker.index, "to be at index but got different worker")
+	}
+}
+
 func makeTxs(size int, b *testing.B) []types.Transactions {
 	key, _ := crypto.GenerateKey()
 

core/block_work.go

@@ -0,0 +1,99 @@
+package core
+
+import (
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/balancer"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/pow"
+)
+
+// nonceResult returns whether the nonce was valid or not.
+type nonceResult struct {
+	index int
+	err   error
+}
+
+// CreateBlockTasks creates new work for the load balancer and returns a
+// channel on which the results of each block is written.
+func CreateBlockTasks(b *balancer.Balancer, blocks []*types.Block, checker pow.PoW) chan nonceResult {
+	if len(blocks) == 0 {
+		return nil
+	}
+
+	var (
+		// Nonce result channel used to post the result of each nonce
+		// checked by the task.
+		nonceResults = make(chan nonceResult, len(blocks))
+		// Error channel (ignored, see below)
+		errch = make(chan error, len(blocks))
+	)
+	for i, block := range blocks {
+		i := i
+		task := balancer.NewTask(func() error {
+			var err error
+			// verify the block's nonce...
+			if !checker.Verify(block) {
+				err = &BlockNonceErr{Hash: block.Hash(), Number: block.Number(), Nonce: block.Nonce()}
+			}
+
+			// ...verify the block uncle's nonces...
+			for _, u := range block.Uncles() {
+				if !checker.Verify(types.NewBlockWithHeader(u)) {
+					err = fmt.Errorf("uncle: %v", BlockNonceErr{Hash: u.Hash(), Number: u.Number, Nonce: block.Nonce()})
+					break
+				}
+			}
+			// ...and write the result on the results chan
+			nonceResults <- nonceResult{i, err}
+			// return nil, ignore error handling by the balancer
+			return nil
+		}, errch)
+		// push task to the balancer
+		b.Push(task)
+		// create transaction tasks for this block
+		CreateTxWork(b, block.Transactions())
+	}
+
+	// we aren't at all interested in the errors
+	// since we handle errors ourself.
+	go func() {
+		// empty out the error channel
+		for i := 0; i < len(blocks); i++ {
+			<-errch
+		}
+		close(errch)
+	}()
+	return nonceResults
+}
+
+// CreateTxWork creates new tasks for the load balancer and derives the public
+// key for each transaction.
+func CreateTxWork(b *balancer.Balancer, txs types.Transactions) {
+	if len(txs) == 0 {
+		return
+	}
+
+	// error channel (ignored, see below)
+	errch := make(chan error, len(txs))
+	// create a new tasks for each transaction and derive the public
+	// key from the sender signature.
+	for i := 0; i < len(txs); i++ {
+		i := i
+		// create new tasks
+		task := balancer.NewTask(func() error {
+			txs[i].FromFrontier()
+			return nil
+		}, errch)
+		// push task to the balancer
+		b.Push(task)
+	}
+
+	// we aren't at all interested in the errors
+	go func() {
+		for i := 0; i < cap(errch); i++ {
+			<-errch
+		}
+		close(errch)
+	}()
+}

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.GOMAXPROCS(0)),
 	}
 	// Seed a fast but crypto originating random generator
 	seed, err := crand.Int(crand.Reader, big.NewInt(math.MaxInt64))
@@ -1120,10 +1123,8 @@ func (self *BlockChain) InsertChain(chain types.Blocks) (int, error) {
 
 		nonceChecked = make([]bool, len(chain))
 	)
-
 	// Start the parallel nonce verifier.
-	nonceAbort, nonceResults := verifyNoncesFromBlocks(self.pow, chain)
-	defer close(nonceAbort)
+	nonceResults := CreateBlockTasks(self.balancer, chain, self.pow)
 
 	txcount := 0
 	for i, block := range chain {
@@ -1138,9 +1139,8 @@ func (self *BlockChain) InsertChain(chain types.Blocks) (int, error) {
 		for !nonceChecked[i] {
 			r := <-nonceResults
 			nonceChecked[r.index] = true
-			if !r.valid {
-				block := chain[r.index]
-				return r.index, &BlockNonceErr{Hash: block.Hash(), Number: block.Number(), Nonce: block.Nonce()}
+			if r.err != nil {
+				return r.index, r.err
 			}
 		}
 
@@ -1252,6 +1252,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)