concurrence init messages from transactions when insert block

update number worker base on numberCPU

make up code

prepare cache signer when insert new block and before addTx to pool

fix some error

fix some error

cmd/tomo/config.go

@@ -151,7 +151,7 @@ func makeConfigNode(ctx *cli.Context) (*node.Node, tomoConfig) {
 		ctx.Set(utils.NATFlag.Name, cfg.NAT)
 	}
 
-	// read passwords from enviroment
+	// read passwords from environment
 	passwords := []string{}
 	for _, env := range cfg.Account.Passwords {
 		if trimmed := strings.TrimSpace(env); trimmed != "" {

core/state_processor.go

@@ -26,6 +26,10 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/params"
 )
+import (
+	"runtime"
+	"sync"
+)
 
 // StateProcessor is a basic Processor, which takes care of transitioning
 // state from one point to another.
@@ -65,6 +69,8 @@ func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg
 	if p.config.DAOForkSupport && p.config.DAOForkBlock != nil && p.config.DAOForkBlock.Cmp(block.Number()) == 0 {
 		misc.ApplyDAOHardFork(statedb)
 	}
+
+	InitSignerInTransactions(p.config, header, block.Transactions())
 	// Iterate over and process the individual transactions
 	for i, tx := range block.Transactions() {
 		statedb.Prepare(tx.Hash(), block.Hash(), i)
@@ -124,3 +130,29 @@ func ApplyTransaction(config *params.ChainConfig, bc *BlockChain, author *common
 
 	return receipt, gas, err
 }
+
+func InitSignerInTransactions(config *params.ChainConfig, header *types.Header, txs types.Transactions) {
+	nWorker := runtime.NumCPU()
+	signer := types.MakeSigner(config, header.Number)
+	chunkSize := txs.Len() / nWorker
+	if txs.Len()%nWorker != 0 {
+		chunkSize++
+	}
+	wg := sync.WaitGroup{}
+	wg.Add(nWorker)
+	for i := 0; i < nWorker; i++ {
+		from := i * chunkSize
+		to := from + chunkSize
+		if to > txs.Len() {
+			to = txs.Len()
+		}
+		go func(from int, to int) {
+			for j := from; j < to; j++ {
+				types.CacheSigner(signer, txs[j])
+				txs[j].CacheHash()
+			}
+			wg.Done()
+		}(from, to)
+	}
+	wg.Wait()
+}

core/tx_pool.go

@@ -793,6 +793,8 @@ func (pool *TxPool) AddRemotes(txs []*types.Transaction) []error {
 
 // addTx enqueues a single transaction into the pool if it is valid.
 func (pool *TxPool) addTx(tx *types.Transaction, local bool) error {
+	tx.CacheHash()
+	types.CacheSigner(pool.signer, tx)
 	pool.mu.Lock()
 	defer pool.mu.Unlock()
 
@@ -811,6 +813,9 @@ func (pool *TxPool) addTx(tx *types.Transaction, local bool) error {
 
 // addTxs attempts to queue a batch of transactions if they are valid.
 func (pool *TxPool) addTxs(txs []*types.Transaction, local bool) []error {
+	for _, tx := range txs {
+		types.CacheSigner(pool.signer, tx)
+	}
 	pool.mu.Lock()
 	defer pool.mu.Unlock()
 

core/types/transaction.go

@@ -206,6 +206,11 @@ func (tx *Transaction) Hash() common.Hash {
 	return v
 }
 
+func (tx *Transaction) CacheHash() {
+	v := rlpHash(tx)
+	tx.hash.Store(v)
+}
+
 // Size returns the true RLP encoded storage size of the transaction, either by
 // encoding and returning it, or returning a previsouly cached value.
 func (tx *Transaction) Size() common.StorageSize {

core/types/transaction_signing.go

@@ -258,3 +258,14 @@ func deriveChainId(v *big.Int) *big.Int {
 	v = new(big.Int).Sub(v, big.NewInt(35))
 	return v.Div(v, big.NewInt(2))
 }
+
+func CacheSigner(signer Signer, tx *Transaction) {
+	if tx == nil {
+		return
+	}
+	addr, err := signer.Sender(tx)
+	if err != nil {
+		return
+	}
+	tx.from.Store(sigCache{signer: signer, from: addr})
+}