From 2cf60ac04074e08bf5f33d72c763e988ea4db82c Mon Sep 17 00:00:00 2001 From: Jimmy Hu Date: Wed, 21 Nov 2018 10:26:36 +0800 Subject: 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 --- core/types/transaction.go | 32 +++++++++++++++++++------------- 1 file changed, 19 insertions(+), 13 deletions(-) (limited to 'core') 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 { -- cgit v1.2.3