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)