From 966326f888daf6d18a6da05814a8aefb430bffa5 Mon Sep 17 00:00:00 2001 From: Jeffrey Wilcke Date: Mon, 22 Feb 2016 23:29:59 +0100 Subject: [PATCH 1/3] 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 | 197 ++++++++++++++++++++++++++++++++++++++ balancer/balancer_test.go | 86 +++++++++++++++++ 2 files changed, 283 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..51843e052a --- /dev/null +++ b/balancer/balancer.go @@ -0,0 +1,197 @@ +package balancer + +import "container/heap" + +const 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) * maxTaskBuffer + // create the new temporary worker + worker := &Worker{ + tasks: make(chan Task, maxTaskBuffer), + temp: true, + start: pending, + pending: pending, + } + 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..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 61ec3107ede38dea1a13a4f117e4ab636ac76325 Mon Sep 17 00:00:00 2001 From: Jeffrey Wilcke Date: Mon, 22 Feb 2016 23:47:06 +0100 Subject: [PATCH 2/3] core: implemented parallel nonce checking & tx pub key derivation The blockchain switched over to the generic tasks handler / load balancer to check for block nonce, uncle nonce and transaction public key gerivation. --- balancer/balancer.go | 5 +- balancer/balancer_test.go | 108 ++++++++++++++++++++++++++++++++++++++ core/block_work.go | 99 ++++++++++++++++++++++++++++++++++ core/blockchain.go | 14 ++--- 4 files changed, 218 insertions(+), 8 deletions(-) create mode 100644 core/block_work.go diff --git a/balancer/balancer.go b/balancer/balancer.go index 51843e052a..f5c8197901 100644 --- a/balancer/balancer.go +++ b/balancer/balancer.go @@ -2,7 +2,7 @@ package balancer import "container/heap" -const maxTaskBuffer = 3000 // maximum amount of work a worker can have in its buffer +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 { @@ -158,13 +158,14 @@ func (b *Balancer) dispatch(task Task) { 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) * maxTaskBuffer + 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 diff --git a/balancer/balancer_test.go b/balancer/balancer_test.go index f645db01c0..2c177bc675 100644 --- a/balancer/balancer_test.go +++ b/balancer/balancer_test.go @@ -1,6 +1,7 @@ package balancer import ( + "container/heap" "math/big" "testing" @@ -9,6 +10,113 @@ import ( "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() 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) From d3ccb8976072b3acdb008e09bd63a02126467928 Mon Sep 17 00:00:00 2001 From: Jeffrey Wilcke Date: Mon, 22 Feb 2016 23:48:27 +0100 Subject: [PATCH 3/3] core: disabled uncle nonce verifier (moved) The verification of uncle nonce has been moved top level blockchain and is now using a parallel approach like the block nonces. --- core/block_validator.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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)) } }