merge conflicts

1 files changed, 98 insertions, 24 deletions

diff --git a/miner/worker.go b/miner/worker.go
index ea8f2e8b5..2b64f3210 100644
--- a/miner/worker.go
+++ b/miner/worker.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"math/big"
 	"sort"
+	"sync"
 
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -41,12 +42,23 @@ func env(block *types.Block, eth *eth.Ethereum) *environment {
 	return env
 }
 
+type Work struct {
+	Number uint64
+	Nonce  []byte
+}
+
 type Agent interface {
-	Comms() chan<- *types.Block
+	Work() chan<- *types.Block
+	SetWorkCh(chan<- Work)
+	Stop()
+	Start()
+	Pow() pow.PoW
 }
 
 type worker struct {
-	agents []chan<- *types.Block
+	mu     sync.Mutex
+	agents []Agent
+	recv   chan Work
 	mux    *event.TypeMux
 	quit   chan struct{}
 	pow    pow.PoW
@@ -57,44 +69,103 @@ type worker struct {
 	coinbase []byte
 
 	current *environment
+
+	mining bool
+}
+
+func newWorker(coinbase []byte, eth *eth.Ethereum) *worker {
+	return &worker{
+		eth:      eth,
+		mux:      eth.EventMux(),
+		recv:     make(chan Work),
+		chain:    eth.ChainManager(),
+		proc:     eth.BlockProcessor(),
+		coinbase: coinbase,
+	}
+}
+
+func (self *worker) start() {
+	self.mining = true
+
+	self.quit = make(chan struct{})
+
+	// spin up agents
+	for _, agent := range self.agents {
+		agent.Start()
+	}
+
+	go self.update()
+	go self.wait()
 }
 
-func (self *worker) register(agent chan<- *types.Block) {
+func (self *worker) stop() {
+	self.mining = false
+
+	close(self.quit)
+}
+
+func (self *worker) register(agent Agent) {
 	self.agents = append(self.agents, agent)
+	agent.SetWorkCh(self.recv)
 }
 
 func (self *worker) update() {
-	events := self.mux.Subscribe(core.NewBlockEvent{}, core.TxPreEvent{}, &LocalTx{})
+	events := self.mux.Subscribe(core.ChainEvent{}, core.TxPreEvent{})
 
 out:
 	for {
 		select {
 		case event := <-events.Chan():
-			switch event := event.(type) {
-			case core.NewBlockEvent:
-				if self.eth.ChainManager().HasBlock(event.Block.Hash()) {
-				}
-			case core.TxPreEvent:
-				if err := self.commitTransaction(event.Tx); err != nil {
-					self.commit()
-				}
+			switch event.(type) {
+			case core.ChainEvent, core.TxPreEvent:
+				self.commitNewWork()
 			}
 		case <-self.quit:
+			// stop all agents
+			for _, agent := range self.agents {
+				agent.Stop()
+			}
 			break out
 		}
 	}
+
+	events.Unsubscribe()
 }
 
-func (self *worker) commit() {
-	self.current.state.Update(ethutil.Big0)
-	self.current.block.SetRoot(self.current.state.Root())
+func (self *worker) wait() {
+	for {
+		for work := range self.recv {
+			block := self.current.block
+			if block.Number().Uint64() == work.Number && block.Nonce() == nil {
+				self.current.block.Header().Nonce = work.Nonce
+
+				if err := self.chain.InsertChain(types.Blocks{self.current.block}); err == nil {
+					self.mux.Post(core.NewMinedBlockEvent{self.current.block})
+				} else {
+					self.commitNewWork()
+				}
+			}
+			break
+		}
+	}
+}
 
-	for _, agent := range self.agents {
-		agent <- self.current.block
+func (self *worker) push() {
+	if self.mining {
+		self.current.block.Header().GasUsed = self.current.totalUsedGas
+		self.current.block.SetRoot(self.current.state.Root())
+
+		// push new work to agents
+		for _, agent := range self.agents {
+			agent.Work() <- self.current.block
+		}
 	}
 }
 
 func (self *worker) commitNewWork() {
+	self.mu.Lock()
+	defer self.mu.Unlock()
+
 	self.current = env(self.chain.NewBlock(self.coinbase), self.eth)
 	parent := self.chain.GetBlock(self.current.block.ParentHash())
 	self.current.coinbase.SetGasPool(core.CalcGasLimit(parent, self.current.block))
@@ -108,19 +179,23 @@ func (self *worker) commitNewWork() {
 		err := self.commitTransaction(tx)
 		switch {
 		case core.IsNonceErr(err):
+			// Remove invalid transactions
 			remove = append(remove, tx)
 		case core.IsGasLimitErr(err):
-			// ignore
-		default:
+			// Break on gas limit
+			break
+		}
+
+		if err != nil {
 			minerlogger.Infoln(err)
-			remove = append(remove, tx)
 		}
 	}
 	self.eth.TxPool().RemoveSet(remove)
 
 	self.current.coinbase.AddAmount(core.BlockReward)
 
-	self.commit()
+	self.current.state.Update(ethutil.Big0)
+	self.push()
 }
 
 var (
@@ -154,14 +229,13 @@ func (self *worker) commitUncle(uncle *types.Header) error {
 
 func (self *worker) commitTransaction(tx *types.Transaction) error {
 	snapshot := self.current.state.Copy()
-	receipt, txGas, err := self.proc.ApplyTransaction(self.current.coinbase, self.current.state, self.current.block, tx, self.current.totalUsedGas, true)
-	if err != nil {
+	receipt, _, err := self.proc.ApplyTransaction(self.current.coinbase, self.current.state, self.current.block, tx, self.current.totalUsedGas, true)
+	if err != nil && (core.IsNonceErr(err) || core.IsGasLimitErr(err)) {
 		self.current.state.Set(snapshot)
 
 		return err
 	}
 
-	self.current.totalUsedGas.Add(self.current.totalUsedGas, txGas)
 	self.current.block.AddTransaction(tx)
 	self.current.block.AddReceipt(receipt)