updates

config/config.go

@@ -699,6 +699,22 @@ type MempoolConfig struct {
 	// Including space needed by encoding (one varint per transaction).
 	// XXX: Unused due to https://github.com/tendermint/tendermint/issues/5796
 	MaxBatchBytes int `mapstructure:"max_batch_bytes"`
+
+	// TTLDuration, if non-zero, defines the maximum amount of time a transaction
+	// can exist for in the mempool.
+	//
+	// Note, if TTLNumBlocks is also defined, a transaction will be removed if it
+	// has existed in the mempool at least TTLNumBlocks number of blocks or if it's
+	// insertion time into the mempool is beyond TTLDuration.
+	TTLDuration time.Duration `mapstructure:"ttl-duration"`
+
+	// TTLNumBlocks, if non-zero, defines the maximum number of blocks a transaction
+	// can exist for in the mempool.
+	//
+	// Note, if TTLDuration is also defined, a transaction will be removed if it
+	// has existed in the mempool at least TTLNumBlocks number of blocks or if
+	// it's insertion time into the mempool is beyond TTLDuration.
+	TTLNumBlocks int64 `mapstructure:"ttl-num-blocks"`
 }
 
 // DefaultMempoolConfig returns a default configuration for the Tendermint mempool
@@ -709,10 +725,12 @@ func DefaultMempoolConfig() *MempoolConfig {
 		WalPath:   "",
 		// Each signature verification takes .5ms, Size reduced until we implement
 		// ABCI Recheck
-		Size:        5000,
-		MaxTxsBytes: 1024 * 1024 * 1024, // 1GB
-		CacheSize:   10000,
-		MaxTxBytes:  1024 * 1024, // 1MB
+		Size:         5000,
+		MaxTxsBytes:  1024 * 1024 * 1024, // 1GB
+		CacheSize:    10000,
+		MaxTxBytes:   1024 * 1024, // 1MB
+		TTLDuration:  0 * time.Second,
+		TTLNumBlocks: 0,
 	}
 }
 

config/toml.go

@@ -371,6 +371,22 @@ max_tx_bytes = {{ .Mempool.MaxTxBytes }}
 # XXX: Unused due to https://github.com/tendermint/tendermint/issues/5796
 max_batch_bytes = {{ .Mempool.MaxBatchBytes }}
 
+# ttl-duration, if non-zero, defines the maximum amount of time a transaction
+# can exist for in the mempool.
+#
+# Note, if ttl-num-blocks is also defined, a transaction will be removed if it
+# has existed in the mempool at least ttl-num-blocks number of blocks or if it's
+# insertion time into the mempool is beyond ttl-duration.
+ttl-duration = "{{ .Mempool.TTLDuration }}"
+
+# ttl-num-blocks, if non-zero, defines the maximum number of blocks a transaction
+# can exist for in the mempool.
+#
+# Note, if ttl-duration is also defined, a transaction will be removed if it
+# has existed in the mempool at least ttl-num-blocks number of blocks or if
+# it's insertion time into the mempool is beyond ttl-duration.
+ttl-num-blocks = {{ .Mempool.TTLNumBlocks }}
+
 #######################################################
 ###         State Sync Configuration Options        ###
 #######################################################

mempool/mempool.go

@@ -66,16 +66,8 @@ type Mempool interface {
 	// Size returns the number of transactions in the mempool.
 	Size() int
 
-	// TxsBytes returns the total size of all txs in the mempool.
-	TxsBytes() int64
-
-	// InitWAL creates a directory for the WAL file and opens a file itself. If
-	// there is an error, it will be of type *PathError.
-	InitWAL() error
-
-	// CloseWAL closes and discards the underlying WAL file.
-	// Any further writes will not be relayed to disk.
-	CloseWAL()
+	// SizeBytes returns the total size of all txs in the mempool.
+	SizeBytes() int64
 }
 
 //--------------------------------------------------------------------------------

mempool/metrics.go

@@ -18,10 +18,25 @@ const (
 type Metrics struct {
 	// Size of the mempool.
 	Size metrics.Gauge
+
 	// Histogram of transaction sizes, in bytes.
 	TxSizeBytes metrics.Histogram
+
 	// Number of failed transactions.
 	FailedTxs metrics.Counter
+
+	// RejectedTxs defines the number of rejected transactions. These are
+	// transactions that passed CheckTx but failed to make it into the mempool
+	// due to resource limits, e.g. mempool is full and no lower priority
+	// transactions exist in the mempool.
+	RejectedTxs metrics.Counter
+
+	// EvictedTxs defines the number of evicted transactions. These are valid
+	// transactions that passed CheckTx and existed in the mempool but were later
+	// evicted to make room for higher priority valid transactions that passed
+	// CheckTx.
+	EvictedTxs metrics.Counter
+
 	// Number of times transactions are rechecked in the mempool.
 	RecheckTimes metrics.Counter
 }
@@ -41,6 +56,7 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 			Name:      "size",
 			Help:      "Size of the mempool (number of uncommitted transactions).",
 		}, labels).With(labelsAndValues...),
+
 		TxSizeBytes: prometheus.NewHistogramFrom(stdprometheus.HistogramOpts{
 			Namespace: namespace,
 			Subsystem: MetricsSubsystem,
@@ -48,12 +64,28 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 			Help:      "Transaction sizes in bytes.",
 			Buckets:   stdprometheus.ExponentialBuckets(1, 3, 17),
 		}, labels).With(labelsAndValues...),
+
 		FailedTxs: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
 			Namespace: namespace,
 			Subsystem: MetricsSubsystem,
 			Name:      "failed_txs",
 			Help:      "Number of failed transactions.",
 		}, labels).With(labelsAndValues...),
+
+		RejectedTxs: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "rejected_txs",
+			Help:      "Number of rejected transactions.",
+		}, labels).With(labelsAndValues...),
+
+		EvictedTxs: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "evicted_txs",
+			Help:      "Number of evicted transactions.",
+		}, labels).With(labelsAndValues...),
+
 		RecheckTimes: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
 			Namespace: namespace,
 			Subsystem: MetricsSubsystem,
@@ -69,6 +101,8 @@ func NopMetrics() *Metrics {
 		Size:         discard.NewGauge(),
 		TxSizeBytes:  discard.NewHistogram(),
 		FailedTxs:    discard.NewCounter(),
+		RejectedTxs:  discard.NewCounter(),
+		EvictedTxs:   discard.NewCounter(),
 		RecheckTimes: discard.NewCounter(),
 	}
 }

mempool/reactor.go

@@ -33,7 +33,7 @@ const (
 type Reactor struct {
 	p2p.BaseReactor
 	config  *cfg.MempoolConfig
-	mempool *CListMempool
+	mempool *TxMempool
 	ids     *mempoolIDs
 }
 
@@ -101,7 +101,7 @@ func newMempoolIDs() *mempoolIDs {
 }
 
 // NewReactor returns a new Reactor with the given config and mempool.
-func NewReactor(config *cfg.MempoolConfig, mempool *CListMempool) *Reactor {
+func NewReactor(config *cfg.MempoolConfig, mempool *TxMempool) *Reactor {
 	memR := &Reactor{
 		config:  config,
 		mempool: mempool,
@@ -205,17 +205,20 @@ func (memR *Reactor) broadcastTxRoutine(peer p2p.Peer) {
 		if !memR.IsRunning() || !peer.IsRunning() {
 			return
 		}
+
 		// This happens because the CElement we were looking at got garbage
 		// collected (removed). That is, .NextWait() returned nil. Go ahead and
 		// start from the beginning.
 		if next == nil {
 			select {
-			case <-memR.mempool.TxsWaitChan(): // Wait until a tx is available
-				if next = memR.mempool.TxsFront(); next == nil {
+			case <-memR.mempool.WaitForNextTx(): // Wait until a tx is available
+				if next = memR.mempool.NextGossipTx(); next == nil {
 					continue
 				}
+
 			case <-peer.Quit():
 				return
+
 			case <-memR.Quit():
 				return
 			}
@@ -242,7 +245,7 @@ func (memR *Reactor) broadcastTxRoutine(peer p2p.Peer) {
 
 		// NOTE: Transaction batching was disabled due to
 		// https://github.com/tendermint/tendermint/issues/5796
-		if _, ok := memTx.senders.Load(peerID); !ok {
+		if ok := memR.mempool.txStore.TxHasPeer(memTx.hash, peerID); !ok {
 			msg := protomem.Message{
 				Sum: &protomem.Message_Txs{
 					Txs: &protomem.Txs{Txs: [][]byte{memTx.tx}},