From 7a7071a7ad57665884888cf1a59039343502cca5 Mon Sep 17 00:00:00 2001
From: Jeffrey Wilcke <geffobscura@gmail.com>
Date: Fri, 19 Feb 2016 20:22:24 +0100
Subject: [PATCH 1/6] balancer: implemented generic load balancer

Balancer is a generic approach to load balancing given any type of
tasks. Tasks are packages that do 1. perform an expensive tasks and
2. yield a good/bad result (i.e. an error).

When tasks are given to the balancer the balancer will selects one of the
least loaded workers and passes on the work and adds it to its work
queue.

The load balancer can be used in all sorts of places and should
generally help with optimisations in places where parallel processing is
useful. The following tasks could be paralellised:

* The transaction pool's validation process whenever it's given a bunch
  of transactions.
* Block & uncle pow verification
* Transaction sender public key derivation
---
 balancer/balancer.go      | 139 ++++++++++++++++++++++++++++++++++++++
 balancer/balancer_test.go |  86 +++++++++++++++++++++++
 2 files changed, 225 insertions(+)
 create mode 100644 balancer/balancer.go
 create mode 100644 balancer/balancer_test.go

diff --git a/balancer/balancer.go b/balancer/balancer.go
new file mode 100644
index 0000000000..8ea39aabc3
--- /dev/null
+++ b/balancer/balancer.go
@@ -0,0 +1,139 @@
+package balancer
+
+import "container/heap"
+
+// Task repsents a single batch of work offered to a worker.
+type Task struct {
+	fn func() error // work function
+	c  chan error   // return channel
+}
+
+// NewTask returns a new task and sets the proper fields.
+func NewTask(fn func() error, c chan error) Task {
+	return Task{
+		fn: fn,
+		c:  c,
+	}
+}
+
+// Worker is a worker that will take one it's assigned tasks
+// and execute it
+type Worker struct {
+	id      int       // worker id
+	tasks   chan Task // tasks to do (buffered)
+	pending int       // count of pending work
+	index   int       // index in the heap
+}
+
+// work will take the oldest task and execute the function and
+// yield the result back in to the return error channel.
+func (w *Worker) work(done chan *Worker) {
+	for {
+		task := <-w.tasks   // get task...
+		task.c <- task.fn() // ...execute the task
+		done <- w           // we're done
+	}
+}
+
+// Pool is a pool of workers and implements containers.Heap
+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) Push(x interface{}) {
+	w := x.(*Worker)  // cast the worker
+	w.index = len(*p) // assign the new index
+
+	*p = append(*p, x.(*Worker))
+}
+
+func (p *Pool) Pop() interface{} {
+	old := *p
+	n := len(old)
+	x := old[n-1]
+	*p = old[0 : n-1]
+	return x
+}
+
+// Balancer is responsible for balancing any given tasks
+// to the pool of workers. The workers are managed by the
+// balancer and will try to make sure that the workers are
+// equally balanced in "work to complete".
+type Balancer struct {
+	pool Pool
+	done chan *Worker
+
+	work chan Task
+}
+
+// New returns a new load balancer
+func New(poolSize int) *Balancer {
+	balancer := &Balancer{
+		done: make(chan *Worker),
+		pool: make(Pool, poolSize),
+		work: make(chan Task),
+	}
+
+	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)}
+		// spawn worker process
+		go func(i int) {
+			operations <- struct{}{}
+			balancer.pool[i].work(balancer.done)
+		}(i)
+	}
+	// spawn own balancer task
+	go balancer.balance(balancer.work)
+
+	// wait for workers to be operations
+	for i := 0; i < poolSize; i++ {
+		<-operations
+	}
+
+	return balancer
+}
+
+// Push pushes the given tasks in to the work channel.
+func (b *Balancer) Push(work Task) {
+	go func() { b.work <- work }()
+}
+
+func (b *Balancer) balance(work chan Task) {
+	for {
+		select {
+		case task := <-work: // get task
+			b.dispatch(task) // dispatch the tasks
+		case w := <-b.done: // worker is done
+			b.completed(w) // handle worker
+		}
+	}
+}
+
+// dispatch dispatches the tasks to the least loaded worker.
+func (b *Balancer) dispatch(task Task) {
+	// Take least loaded worker
+	w := heap.Pop(&b.pool).(*Worker)
+	// send it a task
+	w.tasks <- task
+	// add to its queue
+	w.pending++
+	// put it back in the heap
+	heap.Push(&b.pool, w)
+}
+
+// completed handles the worker and puts it back in the pool
+// based on it's load.
+func (b *Balancer) completed(w *Worker) {
+	// reduce one task
+	w.pending--
+	// remove it from the heap
+	heap.Remove(&b.pool, w.index)
+	// put it back in place
+	heap.Push(&b.pool, w)
+}
diff --git a/balancer/balancer_test.go b/balancer/balancer_test.go
new file mode 100644
index 0000000000..f645db01c0
--- /dev/null
+++ b/balancer/balancer_test.go
@@ -0,0 +1,86 @@
+package balancer
+
+import (
+	"math/big"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
+)
+
+func makeTxs(size int, b *testing.B) []types.Transactions {
+	key, _ := crypto.GenerateKey()
+
+	batches := make([]types.Transactions, b.N)
+	for i := 0; i < b.N; i++ {
+		txs := make(types.Transactions, size)
+		for j := range txs {
+			var err error
+			txs[j], err = types.NewTransaction(0, common.Address{}, new(big.Int), new(big.Int), new(big.Int), nil).SignECDSA(key)
+			if err != nil {
+				b.Fatal(err)
+			}
+		}
+		batches[i] = txs
+	}
+	return batches
+}
+
+const benchTxSize = 1024
+
+func BenchmarkTxsRaw(b *testing.B) {
+	batches := makeTxs(benchTxSize, b)
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		txs := batches[i]
+		b.StartTimer()
+
+		for _, tx := range txs {
+			if _, err := tx.From(); err != nil {
+				b.Fatal(err)
+			}
+		}
+	}
+}
+
+func BenchmarkTxsLoadBalancer(b *testing.B) {
+	balancer := New(4)
+	batches := makeTxs(benchTxSize, b)
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		txs := batches[i]
+
+		var (
+			size      = 8
+			batchsize = len(txs) / size
+			ch        = make(chan error, size)
+		)
+
+		for j := 0; j < size; j++ {
+			j := j
+			task := Task{
+				fn: func() error {
+					for _, tx := range txs[j*size : j*size+batchsize] {
+						if _, err := tx.From(); err != nil {
+							return err
+						}
+					}
+					return nil
+				},
+				c: ch,
+			}
+
+			balancer.Push(task)
+		}
+
+		for j := 0; j < size; j++ {
+			if err := <-ch; err != nil {
+				b.Error(err)
+			}
+		}
+		close(ch)
+	}
+}

From 6e22de0044ea7688967cb3bf9b91fb3f1687752a Mon Sep 17 00:00:00 2001
From: Jeffrey Wilcke <jeffrey@ethereum.org>
Date: Sat, 20 Feb 2016 16:43:30 +0100
Subject: [PATCH 2/6] core: added nonce workers for load balancer

---
 core/block_work.go | 51 ++++++++++++++++++++++++++++++++++++++++++++++
 core/blockchain.go | 11 ++++++++--
 2 files changed, 60 insertions(+), 2 deletions(-)
 create mode 100644 core/block_work.go

diff --git a/core/block_work.go b/core/block_work.go
new file mode 100644
index 0000000000..cadd185308
--- /dev/null
+++ b/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
+}
diff --git a/core/blockchain.go b/core/blockchain.go
index 2c6ff24f97..a5123065f0 100644
--- a/core/blockchain.go
+++ b/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)
-	defer close(nonceAbort)
+	//nonceAbort, nonceResults := verifyNoncesFromBlocks(self.pow, chain)
+	//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)

From a1ab4b34e85efb5a9d4bf003a86aaa3b54d44fc4 Mon Sep 17 00:00:00 2001
From: Jeffrey Wilcke <jeffrey@ethereum.org>
Date: Sat, 20 Feb 2016 18:08:25 +0100
Subject: [PATCH 3/6] quick cehck tx work

---
 core/block_work.go           |  6 ++----
 core/blockchain.go           |  3 +--
 eth/downloader/downloader.go | 39 ++++++++++++++++++++++++++++++++++++
 3 files changed, 42 insertions(+), 6 deletions(-)

diff --git a/core/block_work.go b/core/block_work.go
index cadd185308..353ab83aa1 100644
--- a/core/block_work.go
+++ b/core/block_work.go
@@ -13,7 +13,7 @@ type nonceResult struct {
 	valid bool
 }
 
-func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.PoW) (chan nonceResult, chan struct{}) {
+func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.PoW) chan nonceResult {
 	const workSize = 64
 
 	var (
@@ -36,16 +36,14 @@ func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.P
 		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
+	return nonceResults
 }
diff --git a/core/blockchain.go b/core/blockchain.go
index a5123065f0..bd7d9f300b 100644
--- a/core/blockchain.go
+++ b/core/blockchain.go
@@ -1127,8 +1127,7 @@ func (self *BlockChain) InsertChain(chain types.Blocks) (int, error) {
 	// Start the parallel nonce verifier.
 	//nonceAbort, nonceResults := verifyNoncesFromBlocks(self.pow, chain)
 	//defer close(nonceAbort)
-	nonceResults, donech := balanceBlockWork(self.balancer, chain, self.pow) // ...balance out work
-	defer close(donech)
+	nonceResults := balanceBlockWork(self.balancer, chain, self.pow) // ...balance out work
 
 	txcount := 0
 	for i, block := range chain {
diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 6dad6a2cd6..63ad1b8d25 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -23,11 +23,13 @@ import (
 	"fmt"
 	"math"
 	"math/big"
+	"runtime"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/ethereum/go-ethereum/balancer"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethdb"
@@ -147,6 +149,8 @@ type Downloader struct {
 	cancelCh   chan struct{} // Channel to cancel mid-flight syncs
 	cancelLock sync.RWMutex  // Lock to protect the cancel channel in delivers
 
+	balancer *balancer.Balancer // load balancer to balance out work
+
 	// Testing hooks
 	syncInitHook     func(uint64, uint64)  // Method to call upon initiating a new sync run
 	bodyFetchHook    func([]*types.Header) // Method to call upon starting a block body fetch
@@ -190,6 +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()),
 	}
 }
 
@@ -1516,6 +1521,36 @@ 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 {
@@ -1542,9 +1577,12 @@ func (d *Downloader) process() error {
 			var (
 				blocks   = make([]*types.Block, 0, maxResultsProcess)
 				receipts = make([]types.Receipts, 0, maxResultsProcess)
+				txs      = make(types.Transactions, 0)
 			)
 			items := int(math.Min(float64(len(results)), float64(maxResultsProcess)))
 			for _, result := range results[:items] {
+				txs = append(txs, result.Transactions...)
+
 				switch {
 				case d.mode == FullSync:
 					blocks = append(blocks, types.NewBlockWithHeader(result.Header).WithBody(result.Transactions, result.Uncles))
@@ -1555,6 +1593,7 @@ func (d *Downloader) process() error {
 					}
 				}
 			}
+			balanceTxWork(d.balancer, txs)
 			// Try to process the results, aborting if there's an error
 			var (
 				err   error

From 2ac62c5627df089f2ae465dc41f2e53f87c490b4 Mon Sep 17 00:00:00 2001
From: Jeffrey Wilcke <geffobscura@gmail.com>
Date: Sat, 20 Feb 2016 18:23:42 +0100
Subject: [PATCH 4/6] tmp

---
 balancer/balancer.go |  4 ++--
 core/block_work.go   | 36 +++++++++++++++++++++++++++++++++++-
 core/blockchain.go   |  5 +++++
 3 files changed, 42 insertions(+), 3 deletions(-)

diff --git a/balancer/balancer.go b/balancer/balancer.go
index 8ea39aabc3..3a05726032 100644
--- a/balancer/balancer.go
+++ b/balancer/balancer.go
@@ -107,10 +107,10 @@ func (b *Balancer) Push(work Task) {
 func (b *Balancer) balance(work chan Task) {
 	for {
 		select {
-		case task := <-work: // get task
-			b.dispatch(task) // dispatch the tasks
 		case w := <-b.done: // worker is done
 			b.completed(w) // handle worker
+		case task := <-work: // get task
+			b.dispatch(task) // dispatch the tasks
 		}
 	}
 }
diff --git a/core/block_work.go b/core/block_work.go
index 353ab83aa1..d45559e2a5 100644
--- a/core/block_work.go
+++ b/core/block_work.go
@@ -13,8 +13,42 @@ type nonceResult struct {
 	valid bool
 }
 
+func balanceTxWork(b *balancer.Balancer, txs types.Transactions) {
+	if len(txs) == 0 {
+		return
+	}
+
+	workSize := len(txs) / 4
+
+	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)
+	}()
+}
+
 func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.PoW) chan nonceResult {
-	const workSize = 64
+	workSize := len(blocks) / 4
 
 	var (
 		nonceResults = make(chan nonceResult, len(blocks))                                      // the nonce result channel (buffered)
diff --git a/core/blockchain.go b/core/blockchain.go
index bd7d9f300b..5fca8a1584 100644
--- a/core/blockchain.go
+++ b/core/blockchain.go
@@ -1122,7 +1122,12 @@ 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)

From aae51bd4c39a63ae73eaa0e9e9500225906cfadc Mon Sep 17 00:00:00 2001
From: Jeffrey Wilcke <geffobscura@gmail.com>
Date: Sat, 20 Feb 2016 19:05:39 +0100
Subject: [PATCH 5/6] sep proc

---
 balancer/balancer.go | 14 +++++++++-----
 core/block_work.go   |  5 +++--
 core/blockchain.go   |  2 +-
 3 files changed, 13 insertions(+), 8 deletions(-)

diff --git a/balancer/balancer.go b/balancer/balancer.go
index 3a05726032..709fa91d0a 100644
--- a/balancer/balancer.go
+++ b/balancer/balancer.go
@@ -105,14 +105,18 @@ func (b *Balancer) Push(work Task) {
 }
 
 func (b *Balancer) balance(work chan Task) {
-	for {
-		select {
-		case w := <-b.done: // worker is done
+	go func() {
+		// worker is done
+		for w := range b.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.
diff --git a/core/block_work.go b/core/block_work.go
index d45559e2a5..5eb5e51e6b 100644
--- a/core/block_work.go
+++ b/core/block_work.go
@@ -2,6 +2,7 @@ package core
 
 import (
 	"math"
+	"runtime"
 
 	"github.com/ethereum/go-ethereum/balancer"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -18,7 +19,7 @@ func balanceTxWork(b *balancer.Balancer, txs types.Transactions) {
 		return
 	}
 
-	workSize := len(txs) / 4
+	workSize := len(txs) / runtime.GOMAXPROCS(0)
 
 	errch := make(chan error, int(math.Ceil(float64(len(txs))/float64(workSize)))) // error channel (buffered)
 	for i := 0; i < len(txs); i += workSize {
@@ -48,7 +49,7 @@ func balanceTxWork(b *balancer.Balancer, txs types.Transactions) {
 }
 
 func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.PoW) chan nonceResult {
-	workSize := len(blocks) / 4
+	workSize := len(blocks) / runtime.GOMAXPROCS(0)
 
 	var (
 		nonceResults = make(chan nonceResult, len(blocks))                                      // the nonce result channel (buffered)
diff --git a/core/blockchain.go b/core/blockchain.go
index 5fca8a1584..c21f58b47c 100644
--- a/core/blockchain.go
+++ b/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.NumCPU()),
+		balancer:     balancer.New(runtime.GOMAXPROCS(0)),
 	}
 	// Seed a fast but crypto originating random generator
 	seed, err := crand.Int(crand.Reader, big.NewInt(math.MaxInt64))

From af806346319bbb752a57a01d942df0e26a72313c Mon Sep 17 00:00:00 2001
From: Jeffrey Wilcke <geffobscura@gmail.com>
Date: Sat, 20 Feb 2016 21:05:49 +0100
Subject: [PATCH 6/6] more test

---
 balancer/balancer.go         | 45 ++++++++++++++++++------------------
 core/block_validator.go      |  2 +-
 core/block_work.go           | 26 ++++++++++++++++-----
 core/blockchain.go           | 12 ++--------
 eth/downloader/downloader.go | 36 +++--------------------------
 5 files changed, 48 insertions(+), 73 deletions(-)

diff --git a/balancer/balancer.go b/balancer/balancer.go
index 709fa91d0a..33b0bea21b 100644
--- a/balancer/balancer.go
+++ b/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),
 	}
-
-	operations := make(chan struct{}, poolSize)
-	defer close(operations)
+	heap.Init(&balancer.pool)
 
 	// 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{}{}
-			balancer.pool[i].work(balancer.done)
+			worker.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() {
-		// worker is done
-		for w := range b.done {
+	for {
+		select {
+		case w := <-b.done: // worker is done
 			b.completed(w) // handle worker
-		}
-	}()
-	go func() {
-		// get task
-		for task := range work {
+		case task := <-work: // get task
 			b.dispatch(task) // dispatch the tasks
 		}
-	}()
+	}
 }
 
 // dispatch dispatches the tasks to the least loaded worker.
diff --git a/core/block_validator.go b/core/block_validator.go
index 73c33d8dde..76a6fcdd0d 100644
--- a/core/block_validator.go
+++ b/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))
 		}
 	}
diff --git a/core/block_work.go b/core/block_work.go
index 5eb5e51e6b..390572a6fc 100644
--- a/core/block_work.go
+++ b/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 {
-	workSize := len(blocks) / runtime.GOMAXPROCS(0)
+func BalanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.PoW) chan nonceResult {
+	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)
diff --git a/core/blockchain.go b/core/blockchain.go
index c21f58b47c..9f05451d1f 100644
--- a/core/blockchain.go
+++ b/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)
-	//defer close(nonceAbort)
-	nonceResults := balanceBlockWork(self.balancer, chain, self.pow) // ...balance out work
+	nonceResults := BalanceBlockWork(self.balancer, chain, self.pow)
 
 	txcount := 0
 	for i, block := range chain {
diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 63ad1b8d25..bec1d93b40 100644
--- a/eth/downloader/downloader.go
+++ b/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