fix error transaction underpriced when add sign tx to pool(full)

This commit is contained in:
Nguyen Ba Tam 2018-11-10 11:06:42 +07:00
parent 046ed99c12
commit 4b00409b8f
6 changed files with 50 additions and 29 deletions

View file

@ -216,6 +216,7 @@ type TxPool struct {
wg sync.WaitGroup // for shutdown sync 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 // NewTxPool creates a new transaction pool to gather, sort and filter inbound
@ -593,9 +594,11 @@ func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {
} }
// Drop non-local transactions under our own minimal accepted gas price // 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 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 { if !local && tx.To() != nil && pool.gasPrice.Cmp(tx.GasPrice()) > 0 {
if !tx.IsSpecialTransaction() || (pool.IsMasterNode != nil && !pool.IsMasterNode(from)) {
return ErrUnderpriced return ErrUnderpriced
} }
}
// Ensure the transaction adheres to nonce ordering // Ensure the transaction adheres to nonce ordering
if pool.currentState.GetNonce(from) > tx.Nonce() { if pool.currentState.GetNonce(from) > tx.Nonce() {
return ErrNonceTooLow return ErrNonceTooLow
@ -650,6 +653,10 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
invalidTxCounter.Inc(1) invalidTxCounter.Inc(1)
return false, err 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 the transaction pool is full, discard underpriced transactions
if uint64(len(pool.all)) >= pool.config.GlobalSlots+pool.config.GlobalQueue { if uint64(len(pool.all)) >= pool.config.GlobalSlots+pool.config.GlobalQueue {
// If the new transaction is underpriced, don't accept it // 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()) 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 the transaction is replacing an already pending one, do directly
if list := pool.pending[from]; list != nil && list.Overlaps(tx) { if list := pool.pending[from]; list != nil && list.Overlaps(tx) {
// Nonce already pending, check if required price bump is met // Nonce already pending, check if required price bump is met

View file

@ -579,7 +579,7 @@ func TestTransactionPostponing(t *testing.T) {
txs := []*types.Transaction{} txs := []*types.Transaction{}
for i, key := range keys { for i, key := range keys {
for j := 0; j < 100; j++ { for j := 0; j < 10; j++ {
var tx *types.Transaction var tx *types.Transaction
if (i+j)%2 == 0 { if (i+j)%2 == 0 {
tx = transaction(uint64(j), 25000, key) 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 { 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]) 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 i%2 == 1 {
if _, ok := pool.pending[accs[0]].txs.items[tx.Nonce()]; ok { 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) 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 { if pool.pending[accs[1]] != nil {
t.Errorf("invalidated account still has pending transactions") 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 i%2 == 1 {
if _, ok := pool.queue[accs[1]].txs.items[tx.Nonce()]; !ok { 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) 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)) account, _ := deriveSender(transaction(0, 0, key))
pool.currentState.AddBalance(account, big.NewInt(1000000)) pool.currentState.AddBalance(account, big.NewInt(1000000))
testTxPoolConfig.AccountQueue = 10
// Keep queuing up transactions and make sure all above a limit are dropped // 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 { if err := pool.AddRemote(transaction(i, 100000, key)); err != nil {
t.Fatalf("tx %d: failed to add transaction: %v", i, err) 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 := testTxPoolConfig
config.NoLocals = nolocals 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) 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) pool := NewTxPool(config, params.TestChainConfig, blockchain)
@ -974,8 +975,8 @@ func testTransactionLimitingEquivalency(t *testing.T, origin uint64) {
account1, _ := deriveSender(transaction(0, 0, key1)) account1, _ := deriveSender(transaction(0, 0, key1))
pool1.currentState.AddBalance(account1, big.NewInt(1000000)) pool1.currentState.AddBalance(account1, big.NewInt(1000000))
testTxPoolConfig.AccountQueue = 10
for i := uint64(0); i < testTxPoolConfig.AccountQueue+5; i++ { for i := uint64(0); i < testTxPoolConfig.AccountQueue; i++ {
if err := pool1.AddRemote(transaction(origin+i, 100000, key1)); err != nil { if err := pool1.AddRemote(transaction(origin+i, 100000, key1)); err != nil {
t.Fatalf("tx %d: failed to add transaction: %v", i, err) 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)) pool2.currentState.AddBalance(account2, big.NewInt(1000000))
txs := []*types.Transaction{} 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)) txs = append(txs, transaction(origin+i, 100000, key2))
} }
pool2.AddRemotes(txs) pool2.AddRemotes(txs)
@ -1106,7 +1107,7 @@ func TestTransactionPendingMinimumAllowance(t *testing.T) {
config := testTxPoolConfig config := testTxPoolConfig
config.GlobalSlots = 0 config.GlobalSlots = 0
config.AccountSlots = 5
pool := NewTxPool(config, params.TestChainConfig, blockchain) pool := NewTxPool(config, params.TestChainConfig, blockchain)
defer pool.Stop() defer pool.Stop()
@ -1285,20 +1286,20 @@ func TestTransactionPoolRepricingKeepsLocals(t *testing.T) {
pool.currentState.AddBalance(crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000*1000000)) pool.currentState.AddBalance(crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000*1000000))
} }
// Create transaction (both pending and queued) with a linearly growing gasprice // 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 // Add pending
p_tx := pricedTransaction(i, 100000, big.NewInt(int64(i+1)), keys[2]) p_tx := pricedTransaction(i, 100000, big.NewInt(int64(i+1)), keys[2])
if err := pool.AddLocal(p_tx); err != nil { if err := pool.AddLocal(p_tx); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// Add queued // 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 { if err := pool.AddLocal(q_tx); err != nil {
t.Fatal(err) t.Fatal(err)
} }
} }
pending, queued := pool.Stats() pending, queued := pool.Stats()
expPending, expQueued := 500, 500 expPending, expQueued := 5, 5
validate := func() { validate := func() {
pending, queued = pool.Stats() pending, queued = pool.Stats()
if pending != expPending { if pending != expPending {

View file

@ -404,18 +404,23 @@ type TransactionsByPriceAndNonce struct {
// if after providing it to the constructor. // if after providing it to the constructor.
// It also classifies special txs and normal txs // 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 // Initialize a price based heap with the head transactions
heads := TxByPrice{} heads := TxByPrice{}
specialTxs := Transactions{} specialTxs := Transactions{}
for _, accTxs := range txs { for _, accTxs := range txs {
from, _ := Sender(signer, accTxs[0])
var normalTxs Transactions var normalTxs Transactions
lastSpecialTx := -1 lastSpecialTx := -1
if len(signers) > 0 {
if _, ok := signers[from]; ok {
for i, tx := range accTxs { for i, tx := range accTxs {
if tx.IsSpecialTransaction() { if tx.IsSpecialTransaction() {
lastSpecialTx = i lastSpecialTx = i
} }
} }
}
}
if lastSpecialTx >= 0 { if lastSpecialTx >= 0 {
for i := 0; i <= lastSpecialTx; i++ { for i := 0; i <= lastSpecialTx; i++ {
specialTxs = append(specialTxs, accTxs[i]) specialTxs = append(specialTxs, accTxs[i])
@ -425,10 +430,9 @@ func NewTransactionsByPriceAndNonce(signer Signer, txs map[common.Address]Transa
normalTxs = accTxs normalTxs = accTxs
} }
if len(normalTxs) > 0 { if len(normalTxs) > 0 {
acc, _ := Sender(signer, normalTxs[0])
heads = append(heads, normalTxs[0]) heads = append(heads, normalTxs[0])
// Ensure the sender address is from the signer // Ensure the sender address is from the signer
txs[acc] = normalTxs[1:] txs[from] = normalTxs[1:]
} }
} }
heap.Init(&heads) heap.Init(&heads)

View file

@ -144,7 +144,7 @@ func TestTransactionPriceNonceSort(t *testing.T) {
} }
} }
// Sort the transactions and cross check the nonce ordering // Sort the transactions and cross check the nonce ordering
txset, _ := NewTransactionsByPriceAndNonce(signer, groups) txset, _ := NewTransactionsByPriceAndNonce(signer, groups,nil)
txs := Transactions{} txs := Transactions{}
for tx := txset.Peek(); tx != nil; tx = txset.Peek() { for tx := txset.Peek(); tx != nil; tx = txset.Peek() {

View file

@ -339,6 +339,18 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
} }
return nil 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 return eth, nil

View file

@ -270,7 +270,7 @@ func (self *worker) update() {
self.currentMu.Lock() self.currentMu.Lock()
acc, _ := types.Sender(self.current.signer, ev.Tx) acc, _ := types.Sender(self.current.signer, ev.Tx)
txs := map[common.Address]types.Transactions{acc: {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.current.commitTransactions(self.mux, txset, specialTxs, self.chain, self.coinbase)
self.currentMu.Unlock() self.currentMu.Unlock()
@ -467,7 +467,7 @@ func (self *worker) commitNewWork() {
tstart := time.Now() tstart := time.Now()
parent := self.chain.CurrentBlock() parent := self.chain.CurrentBlock()
var signers map[common.Address]struct{}
// Only try to commit new work if we are mining // Only try to commit new work if we are mining
if atomic.LoadInt32(&self.mining) == 1 { if atomic.LoadInt32(&self.mining) == 1 {
// check if we are right after parent's coinbase in the list // 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) log.Error("Failed when trying to commit new work", "err", err)
return return
} }
signers = snap.Signers
preIndex, curIndex, ok, err := posv.YourTurn(masternodes, snap, parent.Header(), self.coinbase) preIndex, curIndex, ok, err := posv.YourTurn(masternodes, snap, parent.Header(), self.coinbase)
if err != nil { if err != nil {
log.Error("Failed when trying to commit new work", "err", err) 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) log.Error("Failed to fetch pending transactions", "err", err)
return 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) work.commitTransactions(self.mux, txs, specialTxs, self.chain, self.coinbase)
// compute uncles for the new block. // compute uncles for the new block.