mempool: Fix the benchmarks (celestiaorg#934)

* fix mempool benchmarks
* add a benchmark for `mempool:update`

---------

Co-authored-by: Lasaro <[email protected]>

Author: otrack

mempool/bench_test.go

@@ -1,12 +1,20 @@
 package mempool
 
 import (
+	"fmt"
+	"testing"
+
 	"encoding/binary"
 	"sync/atomic"
-	"testing"
 
 	"github.com/cometbft/cometbft/abci/example/kvstore"
+	abciserver "github.com/cometbft/cometbft/abci/server"
+	"github.com/cometbft/cometbft/internal/test"
+	"github.com/cometbft/cometbft/libs/log"
+	cmtrand "github.com/cometbft/cometbft/libs/rand"
 	"github.com/cometbft/cometbft/proxy"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func BenchmarkReap(b *testing.B) {
@@ -15,11 +23,11 @@ func BenchmarkReap(b *testing.B) {
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 
-	mp.config.Size = 100000
+	mp.config.Size = 100_000_000 // so that the nmempool never saturates
 
 	size := 10000
 	for i := 0; i < size; i++ {
-		tx := make([]byte, 8)
+		tx := kvstore.NewTxFromID(i)
 		binary.BigEndian.PutUint64(tx, uint64(i))
 		if err := mp.CheckTx(tx, nil, TxInfo{}); err != nil {
 			b.Fatal(err)
@@ -37,14 +45,12 @@ func BenchmarkCheckTx(b *testing.B) {
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 
-	mp.config.Size = 1000000
+	mp.config.Size = 100_000_000
 
 	b.ResetTimer()
-
 	for i := 0; i < b.N; i++ {
 		b.StopTimer()
-		tx := make([]byte, 8)
-		binary.BigEndian.PutUint64(tx, uint64(i))
+		tx := kvstore.NewTxFromID(i)
 		b.StartTimer()
 
 		if err := mp.CheckTx(tx, nil, TxInfo{}); err != nil {
@@ -59,18 +65,17 @@ func BenchmarkParallelCheckTx(b *testing.B) {
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 
-	mp.config.Size = 100000000
+	mp.config.Size = 100_000_000
 
 	var txcnt uint64
 	next := func() uint64 {
-		return atomic.AddUint64(&txcnt, 1) - 1
+		return atomic.AddUint64(&txcnt, 1)
 	}
 
 	b.ResetTimer()
 	b.RunParallel(func(pb *testing.PB) {
 		for pb.Next() {
-			tx := make([]byte, 8)
-			binary.BigEndian.PutUint64(tx, next())
+			tx := kvstore.NewTxFromID(int(next()))
 			if err := mp.CheckTx(tx, nil, TxInfo{}); err != nil {
 				b.Fatal(err)
 			}
@@ -84,17 +89,60 @@ func BenchmarkCheckDuplicateTx(b *testing.B) {
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 
-	mp.config.Size = 1000000
+	mp.config.Size = 2
 
-	for i := 0; i < b.N; i++ {
-		tx := make([]byte, 8)
-		binary.BigEndian.PutUint64(tx, uint64(i))
-		if err := mp.CheckTx(tx, nil, TxInfo{}); err != nil {
-			b.Fatal(err)
-		}
+	tx := kvstore.NewTxFromID(1)
+	if err := mp.CheckTx(tx, nil, TxInfo{}); err != nil {
+		b.Fatal(err)
+	}
+	e := mp.FlushAppConn()
+	require.True(b, e == nil)
 
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
 		if err := mp.CheckTx(tx, nil, TxInfo{}); err == nil {
 			b.Fatal("tx should be duplicate")
 		}
 	}
+
+}
+
+func BenchmarkUpdateRemoteClient(b *testing.B) {
+	sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", cmtrand.Str(6))
+	app := kvstore.NewInMemoryApplication()
+
+	// Start server
+	server := abciserver.NewSocketServer(sockPath, app)
+	server.SetLogger(log.TestingLogger().With("module", "abci-server"))
+	if err := server.Start(); err != nil {
+		b.Fatalf("Error starting socket server: %v", err.Error())
+	}
+
+	b.Cleanup(func() {
+		if err := server.Stop(); err != nil {
+			b.Error(err)
+		}
+	})
+	cfg := test.ResetTestRoot("mempool_test")
+	mp, cleanup := newMempoolWithAppAndConfig(proxy.NewRemoteClientCreator(sockPath, "socket", true), cfg)
+	defer cleanup()
+
+	b.ResetTimer()
+	for i := 1; i <= b.N; i++ {
+
+		tx := kvstore.NewTxFromID(i)
+
+		e := mp.CheckTx(tx, nil, TxInfo{})
+		require.True(b, e == nil)
+
+		e = mp.FlushAppConn()
+		require.True(b, e == nil)
+
+		require.True(b, mp.Size() == 1)
+
+		var txs = mp.ReapMaxTxs(mp.Size())
+		doCommit(b, mp, app, txs, int64(i))
+		assert.True(b, true)
+	}
+
 }

mempool/cache_test.go

@@ -3,9 +3,10 @@ package mempool
 import (
 	"crypto/rand"
 	"crypto/sha256"
-	"fmt"
 	"testing"
 
+	"fmt"
+
 	"github.com/cometbft/cometbft/abci/example/kvstore"
 	abci "github.com/cometbft/cometbft/abci/types"
 	"github.com/cometbft/cometbft/proxy"

mempool/clist_mempool_test.go

@@ -3,12 +3,13 @@ package mempool
 import (
 	"context"
 	"encoding/binary"
-	"fmt"
 	mrand "math/rand"
 	"os"
 	"testing"
 	"time"
 
+	"fmt"
+
 	"github.com/cosmos/gogoproto/proto"
 	gogotypes "github.com/cosmos/gogoproto/types"
 	"github.com/stretchr/testify/assert"
@@ -741,3 +742,20 @@ func abciResponses(n int, code uint32) []*abci.ExecTxResult {
 	}
 	return responses
 }
+
+func doCommit(t require.TestingT, mp Mempool, app abci.Application, txs types.Txs, height int64) {
+	rfb := &abci.RequestFinalizeBlock{Txs: make([][]byte, len(txs))}
+	for i, tx := range txs {
+		rfb.Txs[i] = tx
+	}
+	_, e := app.FinalizeBlock(context.Background(), rfb)
+	require.True(t, e == nil)
+	mp.Lock()
+	e = mp.FlushAppConn()
+	require.True(t, e == nil)
+	_, e = app.Commit(context.Background(), &abci.RequestCommit{})
+	require.True(t, e == nil)
+	e = mp.Update(height, txs, abciResponses(txs.Len(), abci.CodeTypeOK), nil, nil)
+	require.True(t, e == nil)
+	mp.Unlock()
+}

mempool/mempool.go

@@ -3,9 +3,10 @@ package mempool
 import (
 	"crypto/sha256"
 	"errors"
-	"fmt"
 	"math"
 
+	"fmt"
+
 	abci "github.com/cometbft/cometbft/abci/types"
 	"github.com/cometbft/cometbft/types"
 )

mempool/reactor.go

@@ -2,9 +2,10 @@ package mempool
 
 import (
 	"errors"
-	"fmt"
 	"time"
 
+	"fmt"
+
 	cfg "github.com/cometbft/cometbft/config"
 	"github.com/cometbft/cometbft/libs/clist"
 	"github.com/cometbft/cometbft/libs/log"