diff --git a/balancer/balancer.go b/balancer/balancer.go new file mode 100644 index 0000000000..f5c8197901 --- /dev/null +++ b/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) +} diff --git a/balancer/balancer_test.go b/balancer/balancer_test.go new file mode 100644 index 0000000000..2c177bc675 --- /dev/null +++ b/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) + } +} 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 new file mode 100644 index 0000000000..bf7446a7c9 --- /dev/null +++ b/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) + }() +} diff --git a/core/blockchain.go b/core/blockchain.go index 2c6ff24f97..f5ab863dd1 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.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)