Optimisations and fixed a couple of DDOS issues in the miner

6 files changed, 63 insertions, 43 deletions

diff --git a/core/block_processor.go b/core/block_processor.go
index b4449100f..a9795385f 100644
--- a/core/block_processor.go
+++ b/core/block_processor.go
@@ -73,24 +73,27 @@ func (sm *BlockProcessor) TransitionState(statedb *state.StateDB, parent, block
 	return receipts, nil
 }
 
-func (self *BlockProcessor) ApplyTransaction(coinbase *state.StateObject, state *state.StateDB, block *types.Block, tx *types.Transaction, usedGas *big.Int, transientProcess bool) (*types.Receipt, *big.Int, error) {
+func (self *BlockProcessor) ApplyTransaction(coinbase *state.StateObject, statedb *state.StateDB, block *types.Block, tx *types.Transaction, usedGas *big.Int, transientProcess bool) (*types.Receipt, *big.Int, error) {
 	// If we are mining this block and validating we want to set the logs back to 0
-	state.EmptyLogs()
+	statedb.EmptyLogs()
 
 	txGas := new(big.Int).Set(tx.Gas())
 
-	cb := state.GetStateObject(coinbase.Address())
-	st := NewStateTransition(NewEnv(state, self.bc, tx, block), tx, cb)
+	cb := statedb.GetStateObject(coinbase.Address())
+	st := NewStateTransition(NewEnv(statedb, self.bc, tx, block), tx, cb)
 	_, err := st.TransitionState()
+	if err != nil && (IsNonceErr(err) || state.IsGasLimitErr(err)) {
+		return nil, nil, err
+	}
 
 	txGas.Sub(txGas, st.gas)
 
 	// Update the state with pending changes
-	state.Update(txGas)
+	statedb.Update(txGas)
 
 	cumulative := new(big.Int).Set(usedGas.Add(usedGas, txGas))
-	receipt := types.NewReceipt(state.Root(), cumulative)
-	receipt.SetLogs(state.Logs())
+	receipt := types.NewReceipt(statedb.Root(), cumulative)
+	receipt.SetLogs(statedb.Logs())
 	receipt.Bloom = types.CreateBloom(types.Receipts{receipt})
 	chainlogger.Debugln(receipt)
 
@@ -99,12 +102,12 @@ func (self *BlockProcessor) ApplyTransaction(coinbase *state.StateObject, state
 		go self.eventMux.Post(TxPostEvent{tx})
 	}
 
-	go self.eventMux.Post(state.Logs())
+	go self.eventMux.Post(statedb.Logs())
 
 	return receipt, txGas, err
 }
 
-func (self *BlockProcessor) ApplyTransactions(coinbase *state.StateObject, state *state.StateDB, block *types.Block, txs types.Transactions, transientProcess bool) (types.Receipts, types.Transactions, types.Transactions, types.Transactions, error) {
+func (self *BlockProcessor) ApplyTransactions(coinbase *state.StateObject, statedb *state.StateDB, block *types.Block, txs types.Transactions, transientProcess bool) (types.Receipts, types.Transactions, types.Transactions, types.Transactions, error) {
 	var (
 		receipts           types.Receipts
 		handled, unhandled types.Transactions
@@ -115,12 +118,12 @@ func (self *BlockProcessor) ApplyTransactions(coinbase *state.StateObject, state
 	)
 
 	for _, tx := range txs {
-		receipt, txGas, err := self.ApplyTransaction(coinbase, state, block, tx, totalUsedGas, transientProcess)
+		receipt, txGas, err := self.ApplyTransaction(coinbase, statedb, block, tx, totalUsedGas, transientProcess)
 		if err != nil {
 			switch {
 			case IsNonceErr(err):
 				return nil, nil, nil, nil, err
-			case IsGasLimitErr(err):
+			case state.IsGasLimitErr(err):
 				return nil, nil, nil, nil, err
 			default:
 				statelogger.Infoln(err)
diff --git a/core/chain_manager.go b/core/chain_manager.go
index 286282064..003781791 100644
--- a/core/chain_manager.go
+++ b/core/chain_manager.go
@@ -134,14 +134,11 @@ func (self *ChainManager) State() *state.StateDB {
 func (self *ChainManager) TransState() *state.StateDB {
 	self.tsmu.RLock()
 	defer self.tsmu.RUnlock()
-	//tmp := self.transState
 
 	return self.transState
 }
 
 func (self *ChainManager) setTransState(statedb *state.StateDB) {
-	self.tsmu.Lock()
-	defer self.tsmu.Unlock()
 	self.transState = statedb
 }
 
@@ -361,6 +358,9 @@ func (bc *ChainManager) Stop() {
 }
 
 func (self *ChainManager) InsertChain(chain types.Blocks) error {
+	self.tsmu.Lock()
+	defer self.tsmu.Unlock()
+
 	for _, block := range chain {
 		td, err := self.processor.Process(block)
 		if err != nil {
@@ -376,6 +376,7 @@ func (self *ChainManager) InsertChain(chain types.Blocks) error {
 		}
 		block.Td = td
 
+		var chain, split bool
 		self.mu.Lock()
 		{
 			self.write(block)
@@ -383,16 +384,26 @@ func (self *ChainManager) InsertChain(chain types.Blocks) error {
 			if td.Cmp(self.td) > 0 {
 				if block.Header().Number.Cmp(new(big.Int).Add(cblock.Header().Number, ethutil.Big1)) < 0 {
 					chainlogger.Infof("Split detected. New head #%v (%x) TD=%v, was #%v (%x) TD=%v\n", block.Header().Number, block.Hash()[:4], td, cblock.Header().Number, cblock.Hash()[:4], self.td)
+					split = true
 				}
 
 				self.setTotalDifficulty(td)
 				self.insert(block)
-				self.setTransState(state.New(cblock.Root(), self.db))
 
-				self.eventMux.Post(ChainEvent{block, td})
+				chain = true
 			}
 		}
 		self.mu.Unlock()
+
+		if chain {
+			//self.setTransState(state.New(block.Root(), self.db))
+			self.eventMux.Post(ChainEvent{block, td})
+		}
+
+		if split {
+			self.setTransState(state.New(block.Root(), self.db))
+			self.eventMux.Post(ChainSplitEvent{block})
+		}
 	}
 
 	return nil
@@ -402,3 +413,7 @@ func (self *ChainManager) InsertChain(chain types.Blocks) error {
 func (self *ChainManager) GetAccount(addr []byte) *state.StateObject {
 	return self.State().GetAccount(addr)
 }
+
+func (self *ChainManager) TransMut() *sync.RWMutex {
+	return &self.tsmu
+}
diff --git a/core/error.go b/core/error.go
index 6af48ac2d..e86bacb2d 100644
--- a/core/error.go
+++ b/core/error.go
@@ -68,23 +68,6 @@ func IsValidationErr(err error) bool {
 	return ok
 }
 
-type GasLimitErr struct {
-	Message string
-	Is, Max *big.Int
-}
-
-func IsGasLimitErr(err error) bool {
-	_, ok := err.(*GasLimitErr)
-
-	return ok
-}
-func (err *GasLimitErr) Error() string {
-	return err.Message
-}
-func GasLimitError(is, max *big.Int) *GasLimitErr {
-	return &GasLimitErr{Message: fmt.Sprintf("GasLimit error. Max %s, transaction would take it to %s", max, is), Is: is, Max: max}
-}
-
 type NonceErr struct {
 	Message string
 	Is, Exp uint64
diff --git a/core/events.go b/core/events.go
index fe106da49..4cbbc609c 100644
--- a/core/events.go
+++ b/core/events.go
@@ -13,3 +13,6 @@ type NewBlockEvent struct{ Block *types.Block }
 
 // NewMinedBlockEvent is posted when a block has been imported.
 type NewMinedBlockEvent struct{ Block *types.Block }
+
+// ChainSplit is posted when a new head is detected
+type ChainSplitEvent struct{ Block *types.Block }
diff --git a/core/state_transition.go b/core/state_transition.go
index 33dd45f02..e82be647d 100644
--- a/core/state_transition.go
+++ b/core/state_transition.go
@@ -166,7 +166,8 @@ func (self *StateTransition) TransitionState() (ret []byte, err error) {
 	defer self.RefundGas()
 
 	// Increment the nonce for the next transaction
-	sender.Nonce += 1
+	self.state.SetNonce(sender.Address(), sender.Nonce+1)
+	//sender.Nonce += 1
 
 	// Transaction gas
 	if err = self.UseGas(vm.GasTx); err != nil {
diff --git a/core/transaction_pool.go b/core/transaction_pool.go
index 7a901fcae..894b6c440 100644
--- a/core/transaction_pool.go
+++ b/core/transaction_pool.go
@@ -3,6 +3,7 @@ package core
 import (
 	"errors"
 	"fmt"
+	"sync"
 
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethutil"
@@ -35,6 +36,7 @@ type TxProcessor interface {
 // guarantee a non blocking pool we use a queue channel which can be
 // independently read without needing access to the actual pool.
 type TxPool struct {
+	mu sync.RWMutex
 	// Queueing channel for reading and writing incoming
 	// transactions to
 	queueChan chan *types.Transaction
@@ -97,7 +99,7 @@ func (self *TxPool) addTx(tx *types.Transaction) {
 	self.txs[string(tx.Hash())] = tx
 }
 
-func (self *TxPool) Add(tx *types.Transaction) error {
+func (self *TxPool) add(tx *types.Transaction) error {
 	if self.txs[string(tx.Hash())] != nil {
 		return fmt.Errorf("Known transaction (%x)", tx.Hash()[0:4])
 	}
@@ -128,17 +130,28 @@ func (self *TxPool) Size() int {
 	return len(self.txs)
 }
 
+func (self *TxPool) Add(tx *types.Transaction) error {
+	self.mu.Lock()
+	defer self.mu.Unlock()
+	return self.add(tx)
+}
 func (self *TxPool) AddTransactions(txs []*types.Transaction) {
+	self.mu.Lock()
+	defer self.mu.Unlock()
+
 	for _, tx := range txs {
-		if err := self.Add(tx); err != nil {
-			txplogger.Infoln(err)
+		if err := self.add(tx); err != nil {
+			txplogger.Debugln(err)
 		} else {
-			txplogger.Infof("tx %x\n", tx.Hash()[0:4])
+			txplogger.Debugf("tx %x\n", tx.Hash()[0:4])
 		}
 	}
 }
 
 func (self *TxPool) GetTransactions() (txs types.Transactions) {
+	self.mu.RLock()
+	defer self.mu.RUnlock()
+
 	txs = make(types.Transactions, self.Size())
 	i := 0
 	for _, tx := range self.txs {
@@ -150,30 +163,32 @@ func (self *TxPool) GetTransactions() (txs types.Transactions) {
 }
 
 func (pool *TxPool) RemoveInvalid(query StateQuery) {
+	pool.mu.Lock()
+
 	var removedTxs types.Transactions
 	for _, tx := range pool.txs {
 		sender := query.GetAccount(tx.From())
 		err := pool.ValidateTransaction(tx)
-		fmt.Println(err, sender.Nonce, tx.Nonce())
 		if err != nil || sender.Nonce >= tx.Nonce() {
 			removedTxs = append(removedTxs, tx)
 		}
 	}
+	pool.mu.Unlock()
 
 	pool.RemoveSet(removedTxs)
 }
 
 func (self *TxPool) RemoveSet(txs types.Transactions) {
+	self.mu.Lock()
+	defer self.mu.Unlock()
+
 	for _, tx := range txs {
 		delete(self.txs, string(tx.Hash()))
 	}
 }
 
-func (pool *TxPool) Flush() []*types.Transaction {
-	txList := pool.GetTransactions()
+func (pool *TxPool) Flush() {
 	pool.txs = make(map[string]*types.Transaction)
-
-	return txList
 }
 
 func (pool *TxPool) Start() {