diff --git a/balancer/balancer.go b/balancer/balancer.go new file mode 100644 index 0000000000..33b0bea21b --- /dev/null +++ b/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) +} 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) + } +} 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..390572a6fc --- /dev/null +++ b/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 +} diff --git a/core/blockchain.go b/core/blockchain.go index 2c6ff24f97..9f05451d1f 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.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) - defer close(nonceAbort) + nonceResults := BalanceBlockWork(self.balancer, chain, self.pow) 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) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 6dad6a2cd6..bec1d93b40 100644 --- a/eth/downloader/downloader.go +++ b/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