Skip to content

Commit 0d4bd2d

Browse files
codchenudpatil
codchen
authored and
udpatil
committed
separate limit for pending tx (#202)
1 parent 0bb6253 commit 0d4bd2d

File tree

4 files changed

+91
-14
lines changed

4 files changed

+91
-14
lines changed

‎config/config.go

+14
Original file line numberDiff line numberDiff line change
@@ -800,6 +800,16 @@ type MempoolConfig struct {
800800
// blacklist the peer.
801801
CheckTxErrorBlacklistEnabled bool `mapstructure:"check-tx-error-blacklist-enabled"`
802802
CheckTxErrorThreshold int `mapstructure:"check-tx-error-threshold"`
803+
804+
// Maximum number of transactions in the pending set
805+
PendingSize int `mapstructure:"pending-size"`
806+
807+
// Limit the total size of all txs in the pending set.
808+
MaxPendingTxsBytes int64 `mapstructure:"max-pending-txs-bytes"`
809+
810+
PendingTTLDuration time.Duration `mapstructure:"pending-ttl-duration"`
811+
812+
PendingTTLNumBlocks int64 `mapstructure:"pending-ttl-num-blocks"`
803813
}
804814

805815
// DefaultMempoolConfig returns a default configuration for the Tendermint mempool.
@@ -817,6 +827,10 @@ func DefaultMempoolConfig() *MempoolConfig {
817827
TxNotifyThreshold: 0,
818828
CheckTxErrorBlacklistEnabled: false,
819829
CheckTxErrorThreshold: 0,
830+
PendingSize: 5000,
831+
MaxPendingTxsBytes: 1024 * 1024 * 1024, // 1GB
832+
PendingTTLDuration: 0 * time.Second,
833+
PendingTTLNumBlocks: 0,
820834
}
821835
}
822836

‎config/toml.go

+8
Original file line numberDiff line numberDiff line change
@@ -405,6 +405,14 @@ check-tx-error-blacklist-enabled = {{ .Mempool.CheckTxErrorBlacklistEnabled }}
405405
406406
check-tx-error-threshold = {{ .Mempool.CheckTxErrorThreshold }}
407407
408+
pending-size = {{ .Mempool.PendingSize }}
409+
410+
max-pending-txs-bytes = {{ .Mempool.MaxPendingTxsBytes }}
411+
412+
pending-ttl-duration = {{ .Mempool.PendingTTLDuration }}
413+
414+
pending-ttl-num-blocks = {{ .Mempool.PendingTTLNumBlocks }}
415+
408416
#######################################################
409417
### State Sync Configuration Options ###
410418
#######################################################

‎internal/mempool/mempool.go

+49-14
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,9 @@ type TxMempool struct {
4343
// sizeBytes defines the total size of the mempool (sum of all tx bytes)
4444
sizeBytes int64
4545

46+
// pendingSizeBytes defines the total size of the pending set (sum of all tx bytes)
47+
pendingSizeBytes int64
48+
4649
// cache defines a fixed-size cache of already seen transactions as this
4750
// reduces pressure on the proxyApp.
4851
cache TxCache
@@ -177,9 +180,11 @@ func (txmp *TxMempool) Unlock() {
177180
// Size returns the number of valid transactions in the mempool. It is
178181
// thread-safe.
179182
func (txmp *TxMempool) Size() int {
180-
txSize := txmp.txStore.Size()
181-
pendingSize := txmp.pendingTxs.Size()
182-
return txSize + pendingSize
183+
return txmp.SizeWithoutPending() + txmp.PendingSize()
184+
}
185+
186+
func (txmp *TxMempool) SizeWithoutPending() int {
187+
return txmp.txStore.Size()
183188
}
184189

185190
// PendingSize returns the number of pending transactions in the mempool.
@@ -193,6 +198,10 @@ func (txmp *TxMempool) SizeBytes() int64 {
193198
return atomic.LoadInt64(&txmp.sizeBytes)
194199
}
195200

201+
func (txmp *TxMempool) PendingSizeBytes() int64 {
202+
return atomic.LoadInt64(&txmp.pendingSizeBytes)
203+
}
204+
196205
// FlushAppConn executes FlushSync on the mempool's proxyAppConn.
197206
//
198207
// NOTE: The caller must obtain a write-lock prior to execution.
@@ -326,6 +335,11 @@ func (txmp *TxMempool) CheckTx(
326335
if res.Checker == nil {
327336
return errors.New("no checker available for pending transaction")
328337
}
338+
if err := txmp.canAddPendingTx(wtx); err != nil {
339+
// TODO: eviction strategy for pending transactions
340+
return err
341+
}
342+
atomic.AddInt64(&txmp.pendingSizeBytes, int64(wtx.Size()))
329343
txmp.pendingTxs.Insert(wtx, res, txInfo)
330344
}
331345
}
@@ -410,7 +424,7 @@ func (txmp *TxMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs {
410424
)
411425

412426
var txs []types.Tx
413-
if uint64(txmp.Size()) < txmp.config.TxNotifyThreshold {
427+
if uint64(txmp.SizeWithoutPending()) < txmp.config.TxNotifyThreshold {
414428
// do not reap anything if threshold is not met
415429
return txs
416430
}
@@ -522,7 +536,7 @@ func (txmp *TxMempool) Update(
522536
}
523537
}
524538

525-
txmp.metrics.Size.Set(float64(txmp.Size()))
539+
txmp.metrics.Size.Set(float64(txmp.SizeWithoutPending()))
526540
txmp.metrics.PendingSize.Set(float64(txmp.PendingSize()))
527541
return nil
528542
}
@@ -640,7 +654,7 @@ func (txmp *TxMempool) addNewTransaction(wtx *WrappedTx, res *abci.ResponseCheck
640654
}
641655

642656
txmp.metrics.TxSizeBytes.Observe(float64(wtx.Size()))
643-
txmp.metrics.Size.Set(float64(txmp.Size()))
657+
txmp.metrics.Size.Set(float64(txmp.SizeWithoutPending()))
644658
txmp.metrics.PendingSize.Set(float64(txmp.PendingSize()))
645659

646660
if txmp.insertTx(wtx) {
@@ -649,7 +663,7 @@ func (txmp *TxMempool) addNewTransaction(wtx *WrappedTx, res *abci.ResponseCheck
649663
"priority", wtx.priority,
650664
"tx", fmt.Sprintf("%X", wtx.tx.Hash()),
651665
"height", txmp.height,
652-
"num_txs", txmp.Size(),
666+
"num_txs", txmp.SizeWithoutPending(),
653667
)
654668
txmp.notifyTxsAvailable()
655669
}
@@ -745,12 +759,12 @@ func (txmp *TxMempool) handleRecheckResult(tx types.Tx, res *abci.ResponseCheckT
745759
if txmp.recheckCursor == nil {
746760
txmp.logger.Debug("finished rechecking transactions")
747761

748-
if txmp.Size() > 0 {
762+
if txmp.SizeWithoutPending() > 0 {
749763
txmp.notifyTxsAvailable()
750764
}
751765
}
752766

753-
txmp.metrics.Size.Set(float64(txmp.Size()))
767+
txmp.metrics.Size.Set(float64(txmp.SizeWithoutPending()))
754768
txmp.metrics.PendingSize.Set(float64(txmp.PendingSize()))
755769
}
756770

@@ -803,7 +817,7 @@ func (txmp *TxMempool) updateReCheckTxs(ctx context.Context) {
803817
// the transaction can be inserted into the mempool.
804818
func (txmp *TxMempool) canAddTx(wtx *WrappedTx) error {
805819
var (
806-
numTxs = txmp.Size()
820+
numTxs = txmp.SizeWithoutPending()
807821
sizeBytes = txmp.SizeBytes()
808822
)
809823

@@ -819,6 +833,24 @@ func (txmp *TxMempool) canAddTx(wtx *WrappedTx) error {
819833
return nil
820834
}
821835

836+
func (txmp *TxMempool) canAddPendingTx(wtx *WrappedTx) error {
837+
var (
838+
numTxs = txmp.PendingSize()
839+
sizeBytes = txmp.PendingSizeBytes()
840+
)
841+
842+
if numTxs >= txmp.config.PendingSize || int64(wtx.Size())+sizeBytes > txmp.config.MaxPendingTxsBytes {
843+
return types.ErrMempoolPendingIsFull{
844+
NumTxs: numTxs,
845+
MaxTxs: txmp.config.PendingSize,
846+
TxsBytes: sizeBytes,
847+
MaxTxsBytes: txmp.config.MaxPendingTxsBytes,
848+
}
849+
}
850+
851+
return nil
852+
}
853+
822854
func (txmp *TxMempool) insertTx(wtx *WrappedTx) bool {
823855
if txmp.isInMempool(wtx.tx) {
824856
return false
@@ -935,13 +967,14 @@ func (txmp *TxMempool) purgeExpiredTxs(blockHeight int64) {
935967
}
936968

937969
// remove pending txs that have expired
938-
txmp.pendingTxs.PurgeExpired(txmp.config.TTLNumBlocks, blockHeight, txmp.config.TTLDuration, now, func(wtx *WrappedTx) {
970+
txmp.pendingTxs.PurgeExpired(txmp.config.PendingTTLNumBlocks, blockHeight, txmp.config.PendingTTLDuration, now, func(wtx *WrappedTx) {
971+
atomic.AddInt64(&txmp.pendingSizeBytes, int64(-wtx.Size()))
939972
txmp.expire(blockHeight, wtx)
940973
})
941974
}
942975

943976
func (txmp *TxMempool) notifyTxsAvailable() {
944-
if txmp.Size() == 0 {
977+
if txmp.SizeWithoutPending() == 0 {
945978
return
946979
}
947980

@@ -979,12 +1012,14 @@ func (txmp *TxMempool) AppendCheckTxErr(existingLogs string, log string) string
9791012
func (txmp *TxMempool) handlePendingTransactions() {
9801013
accepted, rejected := txmp.pendingTxs.EvaluatePendingTransactions()
9811014
for _, tx := range accepted {
1015+
atomic.AddInt64(&txmp.pendingSizeBytes, int64(-tx.tx.Size()))
9821016
if err := txmp.addNewTransaction(tx.tx, tx.checkTxResponse.ResponseCheckTx, tx.txInfo); err != nil {
9831017
txmp.logger.Error(fmt.Sprintf("error adding pending transaction: %s", err))
9841018
}
9851019
}
986-
if !txmp.config.KeepInvalidTxsInCache {
987-
for _, tx := range rejected {
1020+
for _, tx := range rejected {
1021+
atomic.AddInt64(&txmp.pendingSizeBytes, int64(-tx.tx.Size()))
1022+
if !txmp.config.KeepInvalidTxsInCache {
9881023
tx.tx.removeHandler(true)
9891024
}
9901025
}

‎types/mempool.go

+20
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"crypto/sha256"
55
"errors"
66
"fmt"
7+
78
tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
89
)
910

@@ -84,6 +85,25 @@ func (e ErrMempoolIsFull) Error() string {
8485
)
8586
}
8687

88+
// ErrMempoolPendingIsFull defines an error where there are too many pending transactions
89+
// not processed yet
90+
type ErrMempoolPendingIsFull struct {
91+
NumTxs int
92+
MaxTxs int
93+
TxsBytes int64
94+
MaxTxsBytes int64
95+
}
96+
97+
func (e ErrMempoolPendingIsFull) Error() string {
98+
return fmt.Sprintf(
99+
"mempool pending set is full: number of txs %d (max: %d), total txs bytes %d (max: %d)",
100+
e.NumTxs,
101+
e.MaxTxs,
102+
e.TxsBytes,
103+
e.MaxTxsBytes,
104+
)
105+
}
106+
87107
// ErrPreCheck defines an error where a transaction fails a pre-check.
88108
type ErrPreCheck struct {
89109
Reason error

0 commit comments

Comments
 (0)