From a1ab4b34e85efb5a9d4bf003a86aaa3b54d44fc4 Mon Sep 17 00:00:00 2001
From: Jeffrey Wilcke <jeffrey@ethereum.org>
Date: Sat, 20 Feb 2016 18:08:25 +0100
Subject: [PATCH] quick cehck tx work

---
 core/block_work.go           |  6 ++----
 core/blockchain.go           |  3 +--
 eth/downloader/downloader.go | 39 ++++++++++++++++++++++++++++++++++++
 3 files changed, 42 insertions(+), 6 deletions(-)

diff --git a/core/block_work.go b/core/block_work.go
index cadd185308..353ab83aa1 100644
--- a/core/block_work.go
+++ b/core/block_work.go
@@ -13,7 +13,7 @@ type nonceResult struct {
 	valid bool
 }
 
-func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.PoW) (chan nonceResult, chan struct{}) {
+func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.PoW) chan nonceResult {
 	const workSize = 64
 
 	var (
@@ -36,16 +36,14 @@ func balanceBlockWork(b *balancer.Balancer, blocks []*types.Block, checker pow.P
 		b.Push(task)
 	}
 
-	donech := make(chan struct{})
 	// we aren't at all interested in the errors
 	// since we handle errors ourself.
 	go func() {
-		<-donech // wait for parent proc to finish
 		for i := 0; i < cap(errch); i++ {
 			<-errch
 		}
 		close(errch)
 	}()
 
-	return nonceResults, donech
+	return nonceResults
 }
diff --git a/core/blockchain.go b/core/blockchain.go
index a5123065f0..bd7d9f300b 100644
--- a/core/blockchain.go
+++ b/core/blockchain.go
@@ -1127,8 +1127,7 @@ func (self *BlockChain) InsertChain(chain types.Blocks) (int, error) {
 	// Start the parallel nonce verifier.
 	//nonceAbort, nonceResults := verifyNoncesFromBlocks(self.pow, chain)
 	//defer close(nonceAbort)
-	nonceResults, donech := balanceBlockWork(self.balancer, chain, self.pow) // ...balance out work
-	defer close(donech)
+	nonceResults := balanceBlockWork(self.balancer, chain, self.pow) // ...balance out work
 
 	txcount := 0
 	for i, block := range chain {
diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 6dad6a2cd6..63ad1b8d25 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -23,11 +23,13 @@ import (
 	"fmt"
 	"math"
 	"math/big"
+	"runtime"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/ethereum/go-ethereum/balancer"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethdb"
@@ -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.New(runtime.NumCPU()),
 	}
 }
 
@@ -1516,6 +1521,36 @@ 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 {
@@ -1542,9 +1577,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 +1593,7 @@ func (d *Downloader) process() error {
 					}
 				}
 			}
+			balanceTxWork(d.balancer, txs)
 			// Try to process the results, aborting if there's an error
 			var (
 				err   error