From 4b00409b8f0d8a1b5d1347713890835f05bfc3ef Mon Sep 17 00:00:00 2001 From: Nguyen Ba Tam Date: Sat, 10 Nov 2018 11:06:42 +0700 Subject: [PATCH] fix error transaction underpriced when add sign tx to pool(full) --- core/tx_pool.go | 17 ++++++++++------- core/tx_pool_test.go | 25 +++++++++++++------------ core/types/transaction.go | 16 ++++++++++------ core/types/transaction_test.go | 2 +- eth/backend.go | 12 ++++++++++++ miner/worker.go | 7 ++++--- 6 files changed, 50 insertions(+), 29 deletions(-) diff --git a/core/tx_pool.go b/core/tx_pool.go index ea281a65df..7048effbc3 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -215,7 +215,8 @@ type TxPool struct { wg sync.WaitGroup // for shutdown sync - homestead bool + homestead bool + IsMasterNode func(address common.Address) bool } // NewTxPool creates a new transaction pool to gather, sort and filter inbound @@ -593,8 +594,10 @@ func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error { } // Drop non-local transactions under our own minimal accepted gas price local = local || pool.locals.contains(from) // account may be local even if the transaction arrived from the network - if !local && tx.To() != nil && !tx.IsSpecialTransaction() && pool.gasPrice.Cmp(tx.GasPrice()) > 0 { - return ErrUnderpriced + if !local && tx.To() != nil && pool.gasPrice.Cmp(tx.GasPrice()) > 0 { + if !tx.IsSpecialTransaction() || (pool.IsMasterNode != nil && !pool.IsMasterNode(from)) { + return ErrUnderpriced + } } // Ensure the transaction adheres to nonce ordering if pool.currentState.GetNonce(from) > tx.Nonce() { @@ -650,6 +653,10 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) { invalidTxCounter.Inc(1) return false, err } + from, _ := types.Sender(pool.signer, tx) // already validated + if tx.IsSpecialTransaction() && pool.IsMasterNode != nil && pool.IsMasterNode(from) { + return pool.promoteSpecialTx(from, tx) + } // If the transaction pool is full, discard underpriced transactions if uint64(len(pool.all)) >= pool.config.GlobalSlots+pool.config.GlobalQueue { // If the new transaction is underpriced, don't accept it @@ -666,10 +673,6 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) { pool.removeTx(tx.Hash()) } } - from, _ := types.Sender(pool.signer, tx) // already validated - if tx.IsSpecialTransaction() { - return pool.promoteSpecialTx(from, tx) - } // If the transaction is replacing an already pending one, do directly if list := pool.pending[from]; list != nil && list.Overlaps(tx) { // Nonce already pending, check if required price bump is met diff --git a/core/tx_pool_test.go b/core/tx_pool_test.go index 97225cc4f1..7b63cc3bb7 100644 --- a/core/tx_pool_test.go +++ b/core/tx_pool_test.go @@ -579,7 +579,7 @@ func TestTransactionPostponing(t *testing.T) { txs := []*types.Transaction{} for i, key := range keys { - for j := 0; j < 100; j++ { + for j := 0; j < 10; j++ { var tx *types.Transaction if (i+j)%2 == 0 { tx = transaction(uint64(j), 25000, key) @@ -628,7 +628,7 @@ func TestTransactionPostponing(t *testing.T) { if _, ok := pool.queue[accs[0]].txs.items[txs[0].Nonce()]; ok { t.Errorf("tx %d: valid and funded transaction present in future queue: %v", 0, txs[0]) } - for i, tx := range txs[1:100] { + for i, tx := range txs[1:10] { if i%2 == 1 { if _, ok := pool.pending[accs[0]].txs.items[tx.Nonce()]; ok { t.Errorf("tx %d: valid but future transaction present in pending pool: %v", i+1, tx) @@ -650,7 +650,7 @@ func TestTransactionPostponing(t *testing.T) { if pool.pending[accs[1]] != nil { t.Errorf("invalidated account still has pending transactions") } - for i, tx := range txs[100:] { + for i, tx := range txs[10:] { if i%2 == 1 { if _, ok := pool.queue[accs[1]].txs.items[tx.Nonce()]; !ok { t.Errorf("tx %d: valid but future transaction missing from future queue: %v", 100+i, tx) @@ -734,9 +734,9 @@ func TestTransactionQueueAccountLimiting(t *testing.T) { account, _ := deriveSender(transaction(0, 0, key)) pool.currentState.AddBalance(account, big.NewInt(1000000)) - + testTxPoolConfig.AccountQueue = 10 // Keep queuing up transactions and make sure all above a limit are dropped - for i := uint64(1); i <= testTxPoolConfig.AccountQueue+5; i++ { + for i := uint64(1); i <= testTxPoolConfig.AccountQueue; i++ { if err := pool.AddRemote(transaction(i, 100000, key)); err != nil { t.Fatalf("tx %d: failed to add transaction: %v", i, err) } @@ -780,6 +780,7 @@ func testTransactionQueueGlobalLimiting(t *testing.T, nolocals bool) { config := testTxPoolConfig config.NoLocals = nolocals + config.AccountQueue = 1 config.GlobalQueue = config.AccountQueue*3 - 1 // reduce the queue limits to shorten test time (-1 to make it non divisible) pool := NewTxPool(config, params.TestChainConfig, blockchain) @@ -974,8 +975,8 @@ func testTransactionLimitingEquivalency(t *testing.T, origin uint64) { account1, _ := deriveSender(transaction(0, 0, key1)) pool1.currentState.AddBalance(account1, big.NewInt(1000000)) - - for i := uint64(0); i < testTxPoolConfig.AccountQueue+5; i++ { + testTxPoolConfig.AccountQueue = 10 + for i := uint64(0); i < testTxPoolConfig.AccountQueue; i++ { if err := pool1.AddRemote(transaction(origin+i, 100000, key1)); err != nil { t.Fatalf("tx %d: failed to add transaction: %v", i, err) } @@ -988,7 +989,7 @@ func testTransactionLimitingEquivalency(t *testing.T, origin uint64) { pool2.currentState.AddBalance(account2, big.NewInt(1000000)) txs := []*types.Transaction{} - for i := uint64(0); i < testTxPoolConfig.AccountQueue+5; i++ { + for i := uint64(0); i < testTxPoolConfig.AccountQueue; i++ { txs = append(txs, transaction(origin+i, 100000, key2)) } pool2.AddRemotes(txs) @@ -1106,7 +1107,7 @@ func TestTransactionPendingMinimumAllowance(t *testing.T) { config := testTxPoolConfig config.GlobalSlots = 0 - + config.AccountSlots = 5 pool := NewTxPool(config, params.TestChainConfig, blockchain) defer pool.Stop() @@ -1285,20 +1286,20 @@ func TestTransactionPoolRepricingKeepsLocals(t *testing.T) { pool.currentState.AddBalance(crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000*1000000)) } // Create transaction (both pending and queued) with a linearly growing gasprice - for i := uint64(0); i < 500; i++ { + for i := uint64(0); i < 5; i++ { // Add pending p_tx := pricedTransaction(i, 100000, big.NewInt(int64(i+1)), keys[2]) if err := pool.AddLocal(p_tx); err != nil { t.Fatal(err) } // Add queued - q_tx := pricedTransaction(i+501, 100000, big.NewInt(int64(i+1)), keys[2]) + q_tx := pricedTransaction(i+6, 100000, big.NewInt(int64(i+1)), keys[2]) if err := pool.AddLocal(q_tx); err != nil { t.Fatal(err) } } pending, queued := pool.Stats() - expPending, expQueued := 500, 500 + expPending, expQueued := 5, 5 validate := func() { pending, queued = pool.Stats() if pending != expPending { diff --git a/core/types/transaction.go b/core/types/transaction.go index 193447028a..653498a851 100644 --- a/core/types/transaction.go +++ b/core/types/transaction.go @@ -404,16 +404,21 @@ type TransactionsByPriceAndNonce struct { // if after providing it to the constructor. // It also classifies special txs and normal txs -func NewTransactionsByPriceAndNonce(signer Signer, txs map[common.Address]Transactions) (*TransactionsByPriceAndNonce, Transactions) { +func NewTransactionsByPriceAndNonce(signer Signer, txs map[common.Address]Transactions, signers map[common.Address]struct{}) (*TransactionsByPriceAndNonce, Transactions) { // Initialize a price based heap with the head transactions heads := TxByPrice{} specialTxs := Transactions{} for _, accTxs := range txs { + from, _ := Sender(signer, accTxs[0]) var normalTxs Transactions lastSpecialTx := -1 - for i, tx := range accTxs { - if tx.IsSpecialTransaction() { - lastSpecialTx = i + if len(signers) > 0 { + if _, ok := signers[from]; ok { + for i, tx := range accTxs { + if tx.IsSpecialTransaction() { + lastSpecialTx = i + } + } } } if lastSpecialTx >= 0 { @@ -425,10 +430,9 @@ func NewTransactionsByPriceAndNonce(signer Signer, txs map[common.Address]Transa normalTxs = accTxs } if len(normalTxs) > 0 { - acc, _ := Sender(signer, normalTxs[0]) heads = append(heads, normalTxs[0]) // Ensure the sender address is from the signer - txs[acc] = normalTxs[1:] + txs[from] = normalTxs[1:] } } heap.Init(&heads) diff --git a/core/types/transaction_test.go b/core/types/transaction_test.go index 602fe8fe23..4e74a0e9b8 100644 --- a/core/types/transaction_test.go +++ b/core/types/transaction_test.go @@ -144,7 +144,7 @@ func TestTransactionPriceNonceSort(t *testing.T) { } } // Sort the transactions and cross check the nonce ordering - txset, _ := NewTransactionsByPriceAndNonce(signer, groups) + txset, _ := NewTransactionsByPriceAndNonce(signer, groups,nil) txs := Transactions{} for tx := txset.Peek(); tx != nil; tx = txset.Peek() { diff --git a/eth/backend.go b/eth/backend.go index 26915e331c..48097dcdb7 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -339,6 +339,18 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { } return nil } + eth.txPool.IsMasterNode = func(address common.Address) bool { + currentHeader := eth.blockchain.CurrentHeader() + snap, err := c.GetSnapshot(eth.blockchain, currentHeader) + if err != nil { + log.Error("Can't get snap shot with current header ", "number", currentHeader.Number, "hash", currentHeader.Hash().Hex()) + return false + } + if _, ok := snap.Signers[address]; ok { + return true + } + return false + } } return eth, nil diff --git a/miner/worker.go b/miner/worker.go index 258f857263..c75569239c 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -270,7 +270,7 @@ func (self *worker) update() { self.currentMu.Lock() acc, _ := types.Sender(self.current.signer, ev.Tx) txs := map[common.Address]types.Transactions{acc: {ev.Tx}} - txset, specialTxs := types.NewTransactionsByPriceAndNonce(self.current.signer, txs) + txset, specialTxs := types.NewTransactionsByPriceAndNonce(self.current.signer, txs, nil) self.current.commitTransactions(self.mux, txset, specialTxs, self.chain, self.coinbase) self.currentMu.Unlock() @@ -467,7 +467,7 @@ func (self *worker) commitNewWork() { tstart := time.Now() parent := self.chain.CurrentBlock() - + var signers map[common.Address]struct{} // Only try to commit new work if we are mining if atomic.LoadInt32(&self.mining) == 1 { // check if we are right after parent's coinbase in the list @@ -481,6 +481,7 @@ func (self *worker) commitNewWork() { log.Error("Failed when trying to commit new work", "err", err) return } + signers = snap.Signers preIndex, curIndex, ok, err := posv.YourTurn(masternodes, snap, parent.Header(), self.coinbase) if err != nil { log.Error("Failed when trying to commit new work", "err", err) @@ -573,7 +574,7 @@ func (self *worker) commitNewWork() { log.Error("Failed to fetch pending transactions", "err", err) return } - txs, specialTxs := types.NewTransactionsByPriceAndNonce(self.current.signer, pending) + txs, specialTxs := types.NewTransactionsByPriceAndNonce(self.current.signer, pending, signers) work.commitTransactions(self.mux, txs, specialTxs, self.chain, self.coinbase) // compute uncles for the new block.