concurrence init messages from transactions when insert block

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,7 +69,9 @@ 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)
 	}
-	// Iterate over and process the individual transactions
+
+	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)
 		receipt, _, err := ApplyTransaction(p.config, p.bc, nil, gp, statedb, header, tx, usedGas, cfg)
@@ -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})
+}