more test

balancer/balancer.go

@@ -1,6 +1,11 @@
 package balancer
 
-import "container/heap"
+import (
+	"container/heap"
+	"runtime"
+)
+
+var B = New(runtime.GOMAXPROCS(0))
 
 // Task repsents a single batch of work offered to a worker.
 type Task struct {
@@ -40,7 +45,12 @@ type Pool []*Worker
 
 func (p Pool) Len() int           { return len(p) }
 func (p Pool) Less(i, j int) bool { return p[i].pending < p[j].pending }
-func (p Pool) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }
+func (p Pool) Swap(i, j int) {
+	p[i].index = j // trade i<->j
+	p[j].index = i // trade j<->i
+
+	p[i], p[j] = p[j], p[i]
+}
 func (p *Pool) Push(x interface{}) {
 	w := x.(*Worker)  // cast the worker
 	w.index = len(*p) // assign the new index
@@ -71,31 +81,24 @@ type Balancer struct {
 func New(poolSize int) *Balancer {
 	balancer := &Balancer{
 		done: make(chan *Worker),
-		pool: make(Pool, poolSize),
 		work: make(chan Task),
+		pool: make(Pool, 0, poolSize),
 	}
-
+	heap.Init(&balancer.pool)
-	operations := make(chan struct{}, poolSize)
-	defer close(operations)
 
 	// fill the pool with the given pool size
 	for i := 0; i < poolSize; i++ {
 		// create new worker
-		balancer.pool[i] = &Worker{id: i, tasks: make(chan Task, 10)}
+		worker := &Worker{id: i, tasks: make(chan Task, 100)}
 		// spawn worker process
 		go func(i int) {
-			operations <- struct{}{}
+			worker.work(balancer.done)
-			balancer.pool[i].work(balancer.done)
 		}(i)
+		heap.Push(&balancer.pool, worker)
 	}
 	// spawn own balancer task
 	go balancer.balance(balancer.work)
 
-	// wait for workers to be operations
-	for i := 0; i < poolSize; i++ {
-		<-operations
-	}
-
 	return balancer
 }
 
@@ -105,18 +108,14 @@ func (b *Balancer) Push(work Task) {
 }
 
 func (b *Balancer) balance(work chan Task) {
-	go func() {
+	for {
-		// worker is done
+		select {
-		for w := range b.done {
+		case w := <-b.done: // worker is done
 			b.completed(w) // handle worker
-		}
+		case task := <-work: // get task
-	}()
-	go func() {
-		// get task
-		for task := range work {
 			b.dispatch(task) // dispatch the tasks
 		}
-	}()
+	}
 }
 
 // dispatch dispatches the tasks to the least loaded worker.

core/block_validator.go

@@ -175,7 +175,7 @@ func (v *BlockValidator) VerifyUncles(block, parent *types.Block) error {
 			return UncleError("uncle[%d](%x)'s parent is not ancestor (%x)", i, hash[:4], uncle.ParentHash[0:4])
 		}
 
-		if err := ValidateHeader(v.Pow, uncle, ancestors[uncle.ParentHash].Header(), true, true); err != nil {
+		if err := ValidateHeader(v.Pow, uncle, ancestors[uncle.ParentHash].Header(), false, true); err != nil {
 			return ValidationError(fmt.Sprintf("uncle[%d](%x) header invalid: %v", i, hash[:4], err))
 		}
 	}

core/block_work.go

@@ -2,7 +2,6 @@ package core
 
 import (
 	"math"
-	"runtime"
 
 	"github.com/ethereum/go-ethereum/balancer"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -14,12 +13,17 @@ type nonceResult struct {
 	valid bool
 }
 
-func balanceTxWork(b *balancer.Balancer, txs types.Transactions) {
+const taskCount = 20
+
+func BalanceTxWork(b *balancer.Balancer, txs types.Transactions) {
 	if len(txs) == 0 {
 		return
 	}
 
-	workSize := len(txs) / runtime.GOMAXPROCS(0)
+	workSize := len(txs) / taskCount
+	if workSize == 0 {
+		workSize = 1
+	}
 
 	errch := make(chan error, int(math.Ceil(float64(len(txs))/float64(workSize)))) // error channel (buffered)
 	for i := 0; i < len(txs); i += workSize {
@@ -48,8 +52,11 @@ func balanceTxWork(b *balancer.Balancer, txs types.Transactions) {
 	}()
 }
 
-func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.PoW) chan nonceResult {
+func BalanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.PoW) chan nonceResult {
-	workSize := len(blocks) / runtime.GOMAXPROCS(0)
+	workSize := len(blocks) / taskCount
+	if workSize == 0 {
+		workSize = 1
+	}
 
 	var (
 		nonceResults = make(chan nonceResult, len(blocks))                                      // the nonce result channel (buffered)
@@ -63,7 +70,14 @@ func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.P
 		// create new tasks
 		task := balancer.NewTask(func() error {
 			for i := 0; i < max-batchNo; i++ {
-				nonceResults <- nonceResult{batchNo + i, checker.Verify(batch[i])}
+				valid := checker.Verify(batch[i])
+				for _, u := range batch[i].Uncles() {
+					if !checker.Verify(types.NewBlockWithHeader(u)) {
+						valid = false
+						break
+					}
+				}
+				nonceResults <- nonceResult{batchNo + i, valid}
 			}
 			return nil
 		}, errch)

core/blockchain.go

@@ -139,7 +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)),
+		balancer:     balancer.B,
 	}
 	// Seed a fast but crypto originating random generator
 	seed, err := crand.Int(crand.Reader, big.NewInt(math.MaxInt64))
@@ -1122,17 +1122,9 @@ func (self *BlockChain) InsertChain(chain types.Blocks) (int, error) {
 		tstart        = time.Now()
 
 		nonceChecked = make([]bool, len(chain))
-		txs          types.Transactions
 	)
-	for _, block := range chain {
-		txs = append(txs, block.Transactions()...)
-	}
-	balanceTxWork(self.balancer, txs)
-
 	// Start the parallel nonce verifier.
-	//nonceAbort, nonceResults := verifyNoncesFromBlocks(self.pow, chain)
+	nonceResults := BalanceBlockWork(self.balancer, chain, self.pow)
-	//defer close(nonceAbort)
-	nonceResults := balanceBlockWork(self.balancer, chain, self.pow) // ...balance out work
 
 	txcount := 0
 	for i, block := range chain {

eth/downloader/downloader.go

@@ -23,7 +23,6 @@ import (
 	"fmt"
 	"math"
 	"math/big"
-	"runtime"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -31,6 +30,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/balancer"
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
@@ -194,7 +194,7 @@ func New(stateDb ethdb.Database, mux *event.TypeMux, hasHeader headerCheckFn, ha
 		bodyWakeCh:       make(chan bool, 1),
 		receiptWakeCh:    make(chan bool, 1),
 		stateWakeCh:      make(chan bool, 1),
-		balancer:         balancer.New(runtime.NumCPU()),
+		balancer:         balancer.B,
 	}
 }
 
@@ -1521,36 +1521,6 @@ func (d *Downloader) fetchParts(errCancel error, deliveryCh chan dataPack, deliv
 	}
 }
 
-func balanceTxWork(b *balancer.Balancer, txs types.Transactions) {
-	const workSize = 64
-
-	errch := make(chan error, int(math.Ceil(float64(len(txs))/float64(workSize)))) // error channel (buffered)
-	for i := 0; i < len(txs); i += workSize {
-		max := int(math.Min(float64(i+workSize), float64(len(txs)))) // get max size...
-		batch := txs[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++ {
-				batch[i].FromFrontier()
-			}
-			return nil
-		}, errch)
-
-		b.Push(task)
-	}
-
-	// we aren't at all interested in the errors
-	// since we handle errors ourself.
-	go func() {
-		for i := 0; i < cap(errch); i++ {
-			<-errch
-		}
-		close(errch)
-	}()
-}
-
 // process takes fetch results from the queue and tries to import them into the
 // chain. The type of import operation will depend on the result contents.
 func (d *Downloader) process() error {
@@ -1593,7 +1563,7 @@ func (d *Downloader) process() error {
 					}
 				}
 			}
-			balanceTxWork(d.balancer, txs)
+			core.BalanceTxWork(d.balancer, txs)
 			// Try to process the results, aborting if there's an error
 			var (
 				err   error