core, dex: Batch process touchSender. Lower priority for tx. (#41)

* dex: Add a tx queue in broadcast
* Modify queue parameter
* Priority select all messages except tx
* Batch process TouchSenders

2 files changed, 20 insertions, 13 deletions

diff --git a/core/types/transaction.go b/core/types/transaction.go
index 63195f4fd..6ee6e4494 100644
--- a/core/types/transaction.go
+++ b/core/types/transaction.go
@@ -21,6 +21,7 @@ import (
 	"errors"
 	"io"
 	"math/big"
+	"runtime"
 	"sync"
 	"sync/atomic"
 
@@ -274,24 +275,29 @@ func (s Transactions) GetRlp(i int) []byte {
 
 // TouchSenders calculates the sender of each transaction and update the cache.
 func (s Transactions) TouchSenders(signer Signer) (errorTx *Transaction, err error) {
+	num := runtime.NumCPU()
+	batchSize := len(s) / num
 	wg := sync.WaitGroup{}
-	wg.Add(len(s))
+	wg.Add(num)
 	txError := make(chan error, 1)
-	for _, tx := range s {
-		go func(tx *Transaction) {
+	for i := 0; i < num; i++ {
+		go func(txs Transactions) {
 			defer wg.Done()
-			if len(txError) > 0 {
-				return
-			}
-			_, err := Sender(signer, tx)
-			if err != nil {
-				select {
-				case txError <- err:
-					errorTx = tx
-				default:
+			for _, tx := range txs {
+				if len(txError) > 0 {
+					return
+				}
+				_, err := Sender(signer, tx)
+				if err != nil {
+					select {
+					case txError <- err:
+						errorTx = tx
+					default:
+					}
+					return
 				}
 			}
-		}(tx)
+		}(s[i*batchSize : (i+1)*batchSize])
 	}
 	wg.Wait()
 	select {
diff --git a/dex/handler.go b/dex/handler.go
index 8ab206d33..6b2e3271b 100644
--- a/dex/handler.go
+++ b/dex/handler.go
@@ -729,6 +729,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			}
 			p.MarkTransaction(tx.Hash())
 		}
+		types.Transactions(txs).TouchSenders(types.MakeSigner(pm.blockchain.Config(), new(big.Int)))
 		pm.txpool.AddRemotes(txs)
 
 	case msg.Code == MetaMsg: