diff --git a/balancer/balancer.go b/balancer/balancer.go index 8ea39aabc3..3a05726032 100644 --- a/balancer/balancer.go +++ b/balancer/balancer.go @@ -107,10 +107,10 @@ func (b *Balancer) Push(work Task) { 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 + case task := <-work: // get task + b.dispatch(task) // dispatch the tasks } } } diff --git a/core/block_work.go b/core/block_work.go index 353ab83aa1..d45559e2a5 100644 --- a/core/block_work.go +++ b/core/block_work.go @@ -13,8 +13,42 @@ type nonceResult struct { valid bool } +func balanceTxWork(b *balancer.Balancer, txs types.Transactions) { + if len(txs) == 0 { + return + } + + workSize := len(txs) / 4 + + 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 { - const workSize = 64 + workSize := len(blocks) / 4 var ( nonceResults = make(chan nonceResult, len(blocks)) // the nonce result channel (buffered) diff --git a/core/blockchain.go b/core/blockchain.go index bd7d9f300b..5fca8a1584 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -1122,7 +1122,12 @@ 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)