feat: added max txs per block to config.toml

config/config.go

@@ -922,6 +922,9 @@ type ConsensusConfig struct {
 	CreateEmptyBlocks         bool          `mapstructure:"create_empty_blocks"`
 	CreateEmptyBlocksInterval time.Duration `mapstructure:"create_empty_blocks_interval"`
 
+	// Max transactions per block. No limit if <= 0.
+	MaxTxs int64 `mapstructure:"max_txs"`
+
 	// Reactor sleep duration parameters
 	PeerGossipSleepDuration     time.Duration `mapstructure:"peer_gossip_sleep_duration"`
 	PeerQueryMaj23SleepDuration time.Duration `mapstructure:"peer_query_maj23_sleep_duration"`

config/toml.go

@@ -454,6 +454,9 @@ skip_timeout_commit = {{ .Consensus.SkipTimeoutCommit }}
 create_empty_blocks = {{ .Consensus.CreateEmptyBlocks }}
 create_empty_blocks_interval = "{{ .Consensus.CreateEmptyBlocksInterval }}"
 
+# Max transactions per block. No limit if <= 0.
+max_txs = {{ .Consensus.MaxTxs }}
+
 # Reactor sleep duration parameters
 peer_gossip_sleep_duration = "{{ .Consensus.PeerGossipSleepDuration }}"
 peer_query_maj23_sleep_duration = "{{ .Consensus.PeerQueryMaj23SleepDuration }}"

consensus/byzantine_test.go

@@ -206,7 +206,7 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 		message := lazyProposer.state.MakeHashMessage(lazyProposer.Round)
 		proof, _ := lazyProposer.privValidator.GenerateVRFProof(message)
 		block, blockParts := lazyProposer.blockExec.CreateProposalBlock(
-			lazyProposer.Height, lazyProposer.state, commit, proposerAddr, lazyProposer.Round, proof,
+			lazyProposer.Height, lazyProposer.state, commit, proposerAddr, lazyProposer.Round, proof, 0,
 		)
 
 		// Flush the WAL. Otherwise, we may not recompute the same proposal to sign,

consensus/common_test.go

@@ -233,7 +233,7 @@ func createProposalBlockSlim(cs *State, vs *validatorStub, round int32) (*types.
 		cs.Logger.Error("enterPropose: Cannot generate vrf proof: %s", err.Error())
 		return nil, nil
 	}
-	return cs.blockExec.CreateProposalBlock(cs.Height, cs.state, commit, proposerAddr, round, proof)
+	return cs.blockExec.CreateProposalBlock(cs.Height, cs.state, commit, proposerAddr, round, proof, 0)
 }
 
 func addVotes(to *State, votes ...*types.Vote) {

consensus/replay_stubs.go

@@ -23,8 +23,9 @@ func (emptyMempool) CheckTxSync(_ types.Tx, _ mempl.TxInfo) (*abci.Response, err
 }
 func (emptyMempool) CheckTxAsync(_ types.Tx, _ mempl.TxInfo, _ func(error), _ func(*abci.Response)) {
 }
-func (emptyMempool) ReapMaxBytesMaxGas(_, _ int64) types.Txs { return types.Txs{} }
-func (emptyMempool) ReapMaxTxs(n int) types.Txs              { return types.Txs{} }
+func (emptyMempool) ReapMaxBytesMaxGas(_, _ int64) types.Txs          { return types.Txs{} }
+func (emptyMempool) ReapMaxBytesMaxGasMaxTxs(_, _, _ int64) types.Txs { return types.Txs{} }
+func (emptyMempool) ReapMaxTxs(n int) types.Txs                       { return types.Txs{} }
 func (emptyMempool) Update(
 	_ *types.Block,
 	_ []*abci.ResponseDeliverTx,

consensus/state.go

@@ -1257,7 +1257,7 @@ func (cs *State) createProposalBlock(round int32) (block *types.Block, blockPart
 		return
 	}
 
-	return cs.blockExec.CreateProposalBlock(cs.Height, cs.state, commit, proposerAddr, round, proof)
+	return cs.blockExec.CreateProposalBlock(cs.Height, cs.state, commit, proposerAddr, round, proof, cs.config.MaxTxs)
 }
 
 // Enter: `timeoutPropose` after entering Propose.

mempool/clist_mempool.go

@@ -610,6 +610,44 @@ func (mem *CListMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs {
 	return txs
 }
 
+// Safe for concurrent use by multiple goroutines.
+func (mem *CListMempool) ReapMaxBytesMaxGasMaxTxs(maxBytes, maxGas, maxTxs int64) types.Txs {
+	mem.updateMtx.RLock()
+	defer mem.updateMtx.RUnlock()
+
+	var totalGas int64
+
+	if maxTxs <= 0 {
+		maxTxs = int64(mem.txs.Len())
+	}
+
+	// TODO: we will get a performance boost if we have a good estimate of avg
+	// size per tx, and set the initial capacity based off of that.
+	// txs := make([]types.Tx, 0, tmmath.MinInt(mem.txs.Len(), max/mem.avgTxSize))
+	txs := make([]types.Tx, 0, mem.txs.Len())
+	protoTxs := tmproto.Data{}
+	for e := mem.txs.Front(); e != nil && len(txs) < int(maxTxs); e = e.Next() {
+		memTx := e.Value.(*mempoolTx)
+
+		protoTxs.Txs = append(protoTxs.Txs, memTx.tx)
+		// Check total size requirement
+		if maxBytes > -1 && int64(protoTxs.Size()) > maxBytes {
+			return txs
+		}
+		// Check total gas requirement.
+		// If maxGas is negative, skip this check.
+		// Since newTotalGas < masGas, which
+		// must be non-negative, it follows that this won't overflow.
+		newTotalGas := totalGas + memTx.gasWanted
+		if maxGas > -1 && newTotalGas > maxGas {
+			return txs
+		}
+		totalGas = newTotalGas
+		txs = append(txs, memTx.tx)
+	}
+	return txs
+}
+
 // Safe for concurrent use by multiple goroutines.
 func (mem *CListMempool) ReapMaxTxs(max int) types.Txs {
 	mem.updateMtx.RLock()

mempool/mempool.go

@@ -25,6 +25,9 @@ type Mempool interface {
 	// transactions (~ all available transactions).
 	ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs
 
+	// Puts cap on txs as well on top of ReapMaxBytesMaxGas
+	ReapMaxBytesMaxGasMaxTxs(maxBytes, maxGas, maxTxs int64) types.Txs
+
 	// ReapMaxTxs reaps up to max transactions from the mempool.
 	// If max is negative, there is no cap on the size of all returned
 	// transactions (~ all available transactions).

mempool/mock/mempool.go

@@ -20,8 +20,9 @@ func (Mempool) CheckTxSync(_ types.Tx, _ mempl.TxInfo) (*abci.Response, error) {
 }
 func (Mempool) CheckTxAsync(_ types.Tx, _ mempl.TxInfo, _ func(error), _ func(*abci.Response)) {
 }
-func (Mempool) ReapMaxBytesMaxGas(_, _ int64) types.Txs { return types.Txs{} }
-func (Mempool) ReapMaxTxs(n int) types.Txs              { return types.Txs{} }
+func (Mempool) ReapMaxBytesMaxGas(_, _ int64) types.Txs          { return types.Txs{} }
+func (Mempool) ReapMaxBytesMaxGasMaxTxs(_, _, _ int64) types.Txs { return types.Txs{} }
+func (Mempool) ReapMaxTxs(n int) types.Txs                       { return types.Txs{} }
 func (Mempool) Update(
 	_ *types.Block,
 	_ []*abci.ResponseDeliverTx,

node/node_test.go

@@ -299,6 +299,7 @@ func TestCreateProposalBlock(t *testing.T) {
 		proposerAddr,
 		0,
 		proof,
+		0,
 	)
 
 	// check that the part set does not exceed the maximum block size
@@ -370,6 +371,7 @@ func TestMaxProposalBlockSize(t *testing.T) {
 		proposerAddr,
 		0,
 		proof,
+		0,
 	)
 
 	pb, err := block.ToProto()

state/execution.go

@@ -125,6 +125,7 @@ func (blockExec *BlockExecutor) CreateProposalBlock(
 	proposerAddr []byte,
 	round int32,
 	proof crypto.Proof,
+	maxTxs int64,
 ) (*types.Block, *types.PartSet) {
 
 	maxBytes := state.ConsensusParams.Block.MaxBytes
@@ -135,7 +136,7 @@ func (blockExec *BlockExecutor) CreateProposalBlock(
 	// Fetch a limited amount of valid txs
 	maxDataBytes := types.MaxDataBytes(maxBytes, commit, evidence)
 
-	txs := blockExec.mempool.ReapMaxBytesMaxGas(maxDataBytes, maxGas)
+	txs := blockExec.mempool.ReapMaxBytesMaxGasMaxTxs(maxDataBytes, maxGas, maxTxs)
 
 	return state.MakeBlock(height, txs, commit, evidence, proposerAddr, round, proof)
 }

test/maverick/consensus/replay_stubs.go

@@ -23,8 +23,9 @@ func (emptyMempool) CheckTxSync(_ types.Tx, _ mempl.TxInfo) (*abci.Response, err
 }
 func (emptyMempool) CheckTxAsync(_ types.Tx, _ mempl.TxInfo, _ func(error), _ func(*abci.Response)) {
 }
-func (emptyMempool) ReapMaxBytesMaxGas(_, _ int64) types.Txs { return types.Txs{} }
-func (emptyMempool) ReapMaxTxs(n int) types.Txs              { return types.Txs{} }
+func (emptyMempool) ReapMaxBytesMaxGas(_, _ int64) types.Txs          { return types.Txs{} }
+func (emptyMempool) ReapMaxBytesMaxGasMaxTxs(_, _, _ int64) types.Txs { return types.Txs{} }
+func (emptyMempool) ReapMaxTxs(n int) types.Txs                       { return types.Txs{} }
 func (emptyMempool) Update(
 	_ *types.Block,
 	_ []*abci.ResponseDeliverTx,

test/maverick/consensus/state.go

@@ -1278,7 +1278,7 @@ func (cs *State) createProposalBlock(round int32) (block *types.Block, blockPart
 		cs.Logger.Error(fmt.Sprintf("enterPropose: %v", err))
 		return
 	}
-	return cs.blockExec.CreateProposalBlock(cs.Height, cs.state, commit, proposerAddr, round, proof)
+	return cs.blockExec.CreateProposalBlock(cs.Height, cs.state, commit, proposerAddr, round, proof, 0)
 }
 
 // Enter: any +2/3 prevotes at next round.