diff --git a/balancer/balancer.go b/balancer/balancer.go index 709fa91d0a..33b0bea21b 100644 --- a/balancer/balancer.go +++ b/balancer/balancer.go @@ -1,6 +1,11 @@ package balancer -import "container/heap" +import ( + "container/heap" + "runtime" +) + +var B = New(runtime.GOMAXPROCS(0)) // Task repsents a single batch of work offered to a worker. type Task struct { @@ -40,7 +45,12 @@ 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], p[j] = p[j], p[i] } +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 @@ -71,31 +81,24 @@ type Balancer struct { func New(poolSize int) *Balancer { balancer := &Balancer{ done: make(chan *Worker), - pool: make(Pool, poolSize), work: make(chan Task), + pool: make(Pool, 0, poolSize), } - - operations := make(chan struct{}, poolSize) - defer close(operations) + heap.Init(&balancer.pool) // fill the pool with the given pool size for i := 0; i < poolSize; i++ { // create new worker - balancer.pool[i] = &Worker{id: i, tasks: make(chan Task, 10)} + worker := &Worker{id: i, tasks: make(chan Task, 100)} // spawn worker process go func(i int) { - operations <- struct{}{} - balancer.pool[i].work(balancer.done) + worker.work(balancer.done) }(i) + heap.Push(&balancer.pool, worker) } // spawn own balancer task go balancer.balance(balancer.work) - // wait for workers to be operations - for i := 0; i < poolSize; i++ { - <-operations - } - return balancer } @@ -105,18 +108,14 @@ func (b *Balancer) Push(work Task) { } func (b *Balancer) balance(work chan Task) { - go func() { - // worker is done - for w := range b.done { + for { + select { + case w := <-b.done: // worker is done b.completed(w) // handle worker - } - }() - go func() { - // get task - for task := range work { + case task := <-work: // get task b.dispatch(task) // dispatch the tasks } - }() + } } // dispatch dispatches the tasks to the least loaded worker. 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 index 5eb5e51e6b..390572a6fc 100644 --- a/core/block_work.go +++ b/core/block_work.go @@ -2,7 +2,6 @@ package core import ( "math" - "runtime" "github.com/ethereum/go-ethereum/balancer" "github.com/ethereum/go-ethereum/core/types" @@ -14,12 +13,17 @@ type nonceResult struct { valid bool } -func balanceTxWork(b *balancer.Balancer, txs types.Transactions) { +const taskCount = 20 + +func BalanceTxWork(b *balancer.Balancer, txs types.Transactions) { if len(txs) == 0 { return } - workSize := len(txs) / runtime.GOMAXPROCS(0) + 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 { @@ -48,8 +52,11 @@ func balanceTxWork(b *balancer.Balancer, txs types.Transactions) { }() } -func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.PoW) chan nonceResult { - workSize := len(blocks) / runtime.GOMAXPROCS(0) +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) @@ -63,7 +70,14 @@ func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.P // create new tasks task := balancer.NewTask(func() error { for i := 0; i < max-batchNo; i++ { - nonceResults <- nonceResult{batchNo + i, checker.Verify(batch[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) diff --git a/core/blockchain.go b/core/blockchain.go index c21f58b47c..9f05451d1f 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -139,7 +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)), + balancer: balancer.B, } // Seed a fast but crypto originating random generator seed, err := crand.Int(crand.Reader, big.NewInt(math.MaxInt64)) @@ -1122,17 +1122,9 @@ func (self *BlockChain) InsertChain(chain types.Blocks) (int, error) { tstart = time.Now() nonceChecked = make([]bool, len(chain)) - txs types.Transactions ) - for _, block := range chain { - txs = append(txs, block.Transactions()...) - } - balanceTxWork(self.balancer, txs) - // Start the parallel nonce verifier. - //nonceAbort, nonceResults := verifyNoncesFromBlocks(self.pow, chain) - //defer close(nonceAbort) - nonceResults := balanceBlockWork(self.balancer, chain, self.pow) // ...balance out work + nonceResults := BalanceBlockWork(self.balancer, chain, self.pow) txcount := 0 for i, block := range chain { diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 63ad1b8d25..bec1d93b40 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -23,7 +23,6 @@ import ( "fmt" "math" "math/big" - "runtime" "strings" "sync" "sync/atomic" @@ -31,6 +30,7 @@ import ( "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" @@ -194,7 +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.New(runtime.NumCPU()), + balancer: balancer.B, } } @@ -1521,36 +1521,6 @@ func (d *Downloader) fetchParts(errCancel error, deliveryCh chan dataPack, deliv } } -func balanceTxWork(b *balancer.Balancer, txs types.Transactions) { - const workSize = 64 - - 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) - }() -} - // process takes fetch results from the queue and tries to import them into the // chain. The type of import operation will depend on the result contents. func (d *Downloader) process() error { @@ -1593,7 +1563,7 @@ func (d *Downloader) process() error { } } } - balanceTxWork(d.balancer, txs) + core.BalanceTxWork(d.balancer, txs) // Try to process the results, aborting if there's an error var ( err error