Merge d3ccb89760 into edd7aa054c

balancer/balancer.go

@@ -0,0 +1,198 @@
+package balancer
+
+import "container/heap"
+
+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 {
+	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
+
+	// work attributes
+	tasks   chan Task     // tasks to do (buffered)
+	pending int           // count of pending work
+	quit    chan struct{} // quit channel
+
+	// heap attributes
+	index int // index in the heap
+
+	// temporary worker attributes
+	temp  bool // is temporary worker
+	start int  // start pending count
+}
+
+// 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 {
+		select {
+		case task := <-w.tasks: // get task...
+			task.c <- task.fn() // ...execute the task
+			done <- w           // we're done
+		case <-w.quit:
+			return
+		}
+	}
+}
+
+// Pool is a pool of workers and implements containers.Heap
+type Pool []*Worker
+
+// Len returns teh length of the pool.
+func (p Pool) Len() int { return len(p) }
+
+// Less returns whether p[i} has less burden than p[j].
+func (p Pool) Less(i, j int) bool { return p[i].pending < p[j].pending }
+
+// Swap swaps p[i] with p[j] and swaps their internal indices.
+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]
+}
+
+// Push pushes the worker x to the pool and sets the index.
+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))
+}
+
+// Pop pops and returns the last item in p.
+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 {
+	poolSize int // initial pool size, important for temp workers.
+	pool     Pool
+	done     chan *Worker
+	work     chan Task
+}
+
+// New returns a new load balancer
+func New(poolSize int) *Balancer {
+	balancer := &Balancer{
+		poolSize: poolSize,
+		done:     make(chan *Worker, poolSize),
+		work:     make(chan Task, poolSize*10),
+		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, maxTaskBuffer)}
+		// spawn worker process
+		go worker.work(balancer.done)
+		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) {
+	b.work <- work
+}
+
+// balance is the main thread of the balancer and handles the dispatching
+// of workers and completion of them.
+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)
+
+	// If a worker is full the next write to the channel will block
+	// we then instead create a new temporary worker that will take
+	// over the task we originally wanted to push on the worker.
+	// What this will do is create a new worker and set the state to be
+	// "temporary". This means that when the "completed" handler is
+	// called we check for temporaryness and discard it if it's empty.
+	// Temporary workers never get precedence over non-temporary
+	// workers by setting their start state to that equal of the
+	// pool size - start pool size * max task buffer.
+	// This may result in false positives with the falsity being that
+	// len(w.tasks) is higher than the actual. This will then create
+	// a temporary worker that may not have been necessary, this is
+	// considered to be a reasonable tradeoff.
+	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 + 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
+	}
+
+	// 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)
+
+	// short circuit if temp worker depleted
+	// results in temp worker being discarded
+	if w.temp && w.pending == w.start {
+		close(w.quit)
+		return
+	}
+	// put it back in place
+	heap.Push(&b.pool, w)
+}

balancer/balancer_test.go

@@ -0,0 +1,194 @@
+package balancer
+
+import (
+	"container/heap"
+	"math/big"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"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()
+
+	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,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)