Merge af80634631 into f8d98f7fcd

balancer/balancer.go

@@ -0,0 +1,142 @@
+package balancer
+
+import (
+	"container/heap"
+	"runtime"
+)
+
+var B = New(runtime.GOMAXPROCS(0))
+
+// 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].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
+
+	*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),
+		work: make(chan Task),
+		pool: make(Pool, 0, poolSize),
+	}
+	heap.Init(&balancer.pool)
+
+	// fill the pool with the given pool size
+	for i := 0; i < poolSize; i++ {
+		// create new worker
+		worker := &Worker{id: i, tasks: make(chan Task, 100)}
+		// spawn worker process
+		go func(i int) {
+			worker.work(balancer.done)
+		}(i)
+		heap.Push(&balancer.pool, worker)
+	}
+	// spawn own balancer task
+	go balancer.balance(balancer.work)
+
+	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 w := <-b.done: // worker is done
+			b.completed(w) // handle worker
+		case task := <-work: // get task
+			b.dispatch(task) // dispatch the tasks
+		}
+	}
+}
+
+// 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)
+}

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)
+	}
+}

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

@@ -0,0 +1,98 @@
+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
+}
+
+const taskCount = 20
+
+func BalanceTxWork(b *balancer.Balancer, txs types.Transactions) {
+	if len(txs) == 0 {
+		return
+	}
+
+	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 {
+		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 {
+	workSize := len(blocks) / taskCount
+	if workSize == 0 {
+		workSize = 1
+	}
+
+	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++ {
+				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)
+
+		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)
+	}()
+
+	return nonceResults
+}

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.B,
 	}
 	// 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)
+	nonceResults := BalanceBlockWork(self.balancer, chain, self.pow)
-	defer close(nonceAbort)
 
 	txcount := 0
 	for i, block := range chain {
@@ -1252,6 +1253,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)

eth/downloader/downloader.go

@@ -28,7 +28,9 @@ import (
 	"sync/atomic"
 	"time"
 
+	"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"
@@ -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.B,
 	}
 }
 
@@ -1542,9 +1547,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 +1563,7 @@ func (d *Downloader) process() error {
 					}
 				}
 			}
+			core.BalanceTxWork(d.balancer, txs)
 			// Try to process the results, aborting if there's an error
 			var (
 				err   error