Perf: Increase buffer size for pubsub server to boost performance (#167)

* Increase buffer size for pubsub server

* Add more timeout for test failure

* Add more timeout

* Fix test split scripts

* Fix test split

* Fix unit test

* Unit test

* Unit test

.github/workflows/tests.yml

@@ -38,7 +38,8 @@ jobs:
       - name: Run Go Tests
         run: |
           NUM_SPLIT=20
-          make test-group-${{matrix.part}} NUM_SPLIT=20
+          make split-test-packages
+          make test-group-${{matrix.part}}
 
       - name: Upload coverage artifact
         uses: actions/upload-artifact@v3
@@ -99,7 +100,7 @@ jobs:
         with:
           max_attempts: 2
           retry_on: error
-          timeout_seconds: 30
+          timeout_seconds: 60
           command: |
               jobs=$(curl https://api.github.com/repos/${{ github.repository }}/actions/runs/${{ github.run_id }}/jobs)
               job_statuses=$(echo "$jobs" | jq -r '.jobs[] | .conclusion')

Makefile

@@ -345,6 +345,6 @@ $(BUILDDIR)/packages.txt:$(GO_TEST_FILES) $(BUILDDIR)
 	go list -f "{{ if (or .TestGoFiles .XTestGoFiles) }}{{ .ImportPath }}{{ end }}" ./... | sort > $@
 
 split-test-packages:$(BUILDDIR)/packages.txt
-	split -d -n l/$(NUM_SPLIT) $< $<.
+	split -d -l $(NUM_SPLIT) $< $<.
 test-group-%:split-test-packages
-	cat $(BUILDDIR)/packages.txt.$* | xargs go test -mod=readonly -timeout=10m -race -covermode=atomic -coverprofile=$*.profile.out
+	cat $(BUILDDIR)/packages.txt.$* | xargs go test -mod=readonly -timeout=15m -race -covermode=atomic -coverprofile=$*.profile.out

internal/consensus/mempool_test.go

@@ -176,8 +176,8 @@ func TestMempoolTxConcurrentWithCommit(t *testing.T) {
 			headerEvent := msg.Data().(types.EventDataNewBlockHeader)
 			n += headerEvent.NumTxs
 			logger.Info("new transactions", "nTxs", headerEvent.NumTxs, "total", n)
-		case <-time.After(30 * time.Second):
-			t.Fatal("Timed out waiting 30s to commit blocks with transactions")
+		case <-time.After(60 * time.Second):
+			t.Fatal("Timed out waiting 60s to commit blocks with transactions")
 		}
 	}
 }

internal/consensus/reactor_test.go

@@ -854,8 +854,8 @@ func TestReactorVotingPowerChange(t *testing.T) {
 		func() bool {
 			return previousTotalVotingPower != states[0].GetRoundState().LastValidators.TotalVotingPower()
 		},
-		5*time.Second,
-		300*time.Millisecond,
+		30*time.Second,
+		100*time.Millisecond,
 		"expected voting power to change (before: %d, after: %d)",
 		previousTotalVotingPower,
 		states[0].GetRoundState().LastValidators.TotalVotingPower(),
@@ -875,8 +875,8 @@ func TestReactorVotingPowerChange(t *testing.T) {
 		func() bool {
 			return previousTotalVotingPower != states[0].GetRoundState().LastValidators.TotalVotingPower()
 		},
-		5*time.Second,
-		300*time.Millisecond,
+		30*time.Second,
+		100*time.Millisecond,
 		"expected voting power to change (before: %d, after: %d)",
 		previousTotalVotingPower,
 		states[0].GetRoundState().LastValidators.TotalVotingPower(),
@@ -895,8 +895,8 @@ func TestReactorVotingPowerChange(t *testing.T) {
 		func() bool {
 			return previousTotalVotingPower != states[0].GetRoundState().LastValidators.TotalVotingPower()
 		},
-		5*time.Second,
-		300*time.Millisecond,
+		30*time.Second,
+		100*time.Millisecond,
 		"expected voting power to change (before: %d, after: %d)",
 		previousTotalVotingPower,
 		states[0].GetRoundState().LastValidators.TotalVotingPower(),

internal/consensus/state_test.go

@@ -2482,7 +2482,6 @@ func TestWaitingTimeoutProposeOnNewRound(t *testing.T) {
 
 	round++ // moving to the next round
 	ensureNewRound(t, newRoundCh, height, round)
-
 	rs := cs1.GetRoundState()
 	assert.Equal(t, rs.Step, cstypes.RoundStepPropose) // P0 does not prevote before timeoutPropose expires
 

internal/eventbus/event_bus.go

@@ -13,6 +13,8 @@ import (
 	"github.com/tendermint/tendermint/types"
 )
 
+var DefaultBufferCapacity = 100
+
 // Subscription is a proxy interface for a pubsub Subscription.
 type Subscription interface {
 	ID() string
@@ -30,7 +32,7 @@ type EventBus struct {
 // NewDefault returns a new event bus with default options.
 func NewDefault(l log.Logger) *EventBus {
 	logger := l.With("module", "eventbus")
-	pubsub := tmpubsub.NewServer(l, tmpubsub.BufferCapacity(0))
+	pubsub := tmpubsub.NewServer(l, tmpubsub.BufferCapacity(DefaultBufferCapacity))
 	b := &EventBus{pubsub: pubsub}
 	b.BaseService = *service.NewBaseService(logger, "EventBus", b)
 	return b