chore: initialise the mock reactor as part of the node setup

Author: rach-id

node/node.go

@@ -5,6 +5,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/tendermint/tendermint/p2p/load"
+	"github.com/tendermint/tendermint/p2p/mock"
 	"net"
 	"net/http"
 	"strings"
@@ -237,6 +239,7 @@ type Node struct {
 	tracer            trace.Tracer
 	pyroscopeProfiler *pyroscope.Profiler
 	pyroscopeTracer   *sdktrace.TracerProvider
+	mockReactor       *mock.Reactor
 }
 
 func initDBs(config *cfg.Config, dbProvider DBProvider) (blockStore *store.BlockStore, stateDB dbm.DB, err error) {
@@ -623,6 +626,7 @@ func createSwitch(config *cfg.Config,
 	stateSyncReactor *statesync.Reactor,
 	consensusReactor *cs.Reactor,
 	evidenceReactor *evidence.Reactor,
+	mockReactor *load.MockReactor,
 	nodeInfo p2p.NodeInfo,
 	nodeKey *p2p.NodeKey,
 	p2pLogger log.Logger,
@@ -641,6 +645,7 @@ func createSwitch(config *cfg.Config,
 	sw.AddReactor("CONSENSUS", consensusReactor)
 	sw.AddReactor("EVIDENCE", evidenceReactor)
 	sw.AddReactor("STATESYNC", stateSyncReactor)
+	sw.AddReactor("MOCK", mockReactor)
 
 	sw.SetNodeInfo(nodeInfo)
 	sw.SetNodeKey(nodeKey)
@@ -927,11 +932,59 @@ func NewNode(config *cfg.Config,
 	// Setup Transport.
 	transport, peerFilters := createTransport(config, nodeInfo, nodeKey, proxyApp, tracer)
 
+	mockReactor := load.NewMockReactor(load.DefaultTestChannels, 100)
+
+	go func() {
+		for {
+			time.Sleep(10 * time.Second)
+			mockReactor.PrintReceiveSpeed()
+		}
+	}()
+
+	go func() {
+		mockReactor.FloodAllPeers(10*time.Minute,
+			load.FirstChannel,
+			load.SecondChannel,
+			load.ThirdChannel,
+			load.FourthChannel,
+			load.FifthChannel,
+			load.SixthChannel,
+			load.SeventhChannel,
+			load.EighthChannel,
+			load.NinthChannel,
+			load.TenthChannel,
+		)
+	}()
+
+	go func() {
+		for _, size := range []int{
+			500,
+			1_000,
+			5_000,
+			10_000,
+			50_000,
+			100_000,
+			500_000,
+			1_000_000,
+			5_000_000,
+			10_000_000,
+			20_000_000,
+			30_000_000,
+			50_000_000,
+			100_000_000,
+			200_000_000,
+		} {
+			time.Sleep(30 * time.Second)
+			mockReactor.IncreaseSize(int64(size))
+			logger.Error("======> increased flood size", "size", size)
+		}
+	}()
+
 	// Setup Switch.
 	p2pLogger := logger.With("module", "p2p")
 	sw := createSwitch(
 		config, transport, p2pMetrics, peerFilters, mempoolReactor, bcReactor,
-		stateSyncReactor, consensusReactor, evidenceReactor, nodeInfo, nodeKey, p2pLogger, tracer,
+		stateSyncReactor, consensusReactor, evidenceReactor, mockReactor, nodeInfo, nodeKey, p2pLogger, tracer,
 	)
 
 	err = sw.AddPersistentPeers(splitAndTrimEmpty(config.P2P.PersistentPeers, ",", " "))
@@ -1368,6 +1421,11 @@ func (n *Node) ConsensusReactor() *cs.Reactor {
 	return n.consensusReactor
 }
 
+// ConsensusReactor returns the Node's ConsensusReactor.
+func (n *Node) MockReactor() *mock.Reactor {
+	return n.mockReactor
+}
+
 // MempoolReactor returns the Node's mempool reactor.
 func (n *Node) MempoolReactor() p2p.Reactor {
 	return n.mempoolReactor

p2p/conn/connection.go

@@ -41,8 +41,8 @@ const (
 	defaultFlushThrottle = 100 * time.Millisecond
 
 	defaultSendQueueCapacity   = 1
-	defaultRecvBufferCapacity  = 4096
-	defaultRecvMessageCapacity = 22020096      // 21MB
+	defaultRecvBufferCapacity  = 40960000
+	defaultRecvMessageCapacity = 22020096000
 	defaultSendRate            = int64(512000) // 500KB/s
 	defaultRecvRate            = int64(512000) // 500KB/s
 	defaultSendTimeout         = 10 * time.Second

p2p/load/mock_reactor.go

@@ -30,12 +30,12 @@ const (
 var priorities = make(map[byte]int)
 
 func init() {
-	for _, ch := range defaultTestChannels {
+	for _, ch := range DefaultTestChannels {
 		priorities[ch.ID] = ch.Priority
 	}
 }
 
-var defaultTestChannels = []*p2p.ChannelDescriptor{
+var DefaultTestChannels = []*p2p.ChannelDescriptor{
 	{
 		ID:                  FirstChannel,
 		Priority:            1,
@@ -135,7 +135,6 @@ var defaultMsgSizes = []int{
 type MockReactor struct {
 	p2p.BaseReactor
 	channels []*conn.ChannelDescriptor
-	sizes    map[byte]int
 
 	mtx                     sync.Mutex
 	peers                   map[p2p.ID]p2p.Peer
@@ -149,21 +148,17 @@ type MockReactor struct {
 }
 
 // NewMockReactor creates a new mock reactor.
-func NewMockReactor(channels []*conn.ChannelDescriptor, msgSizes []int) *MockReactor {
+func NewMockReactor(channels []*conn.ChannelDescriptor, msgSize int) *MockReactor {
 	s := atomic.Int64{}
-	s.Store(200)
+	s.Store(int64(msgSize))
 	mr := &MockReactor{
 		channels:                channels,
 		peers:                   make(map[p2p.ID]p2p.Peer),
-		sizes:                   make(map[byte]int),
 		startTime:               map[string]time.Time{},
 		speed:                   map[string]float64{},
 		cumulativeReceivedBytes: map[string]int{},
 		size:                    s,
 	}
-	for i, ch := range channels {
-		mr.sizes[ch.ID] = msgSizes[i]
-	}
 	mr.BaseReactor = *p2p.NewBaseReactor("MockReactor", mr)
 	return mr
 }
@@ -292,22 +287,19 @@ func (mr *MockReactor) FillChannel(id p2p.ID, chID byte, count, msgSize int) (bo
 	return true, count, end.Sub(start)
 }
 
-func (mr *MockReactor) FloodChannel(wg *sync.WaitGroup, id p2p.ID, d time.Duration, chIDs ...byte) {
+func (mr *MockReactor) FloodChannel(id p2p.ID, d time.Duration, chIDs ...byte) {
 	for _, chID := range chIDs {
-		wg.Add(1)
-		size := mr.sizes[chID]
-		go func(d time.Duration, chID byte, size int) {
+		go func(d time.Duration, chID byte) {
 			start := time.Now()
-			defer wg.Done()
 			for time.Since(start) < d {
 				mr.SendBytes(id, chID)
 			}
-		}(d, chID, size)
+		}(d, chID)
 	}
 }
 
-func (mr *MockReactor) FloodAllPeers(wg *sync.WaitGroup, d time.Duration, chIDs ...byte) {
+func (mr *MockReactor) FloodAllPeers(d time.Duration, chIDs ...byte) {
 	for _, peer := range mr.peers {
-		mr.FloodChannel(wg, peer.ID(), d, chIDs...)
+		mr.FloodChannel(peer.ID(), d, chIDs...)
 	}
 }

p2p/load/multi_conn_test.go

@@ -27,7 +27,7 @@ func TestMultipleConnections(t *testing.T) {
 	chainID := "base-30"
 
 	for i := 0; i < peerCount; i++ {
-		reactor := NewMockReactor(defaultTestChannels, defaultMsgSizes)
+		reactor := NewMockReactor(DefaultTestChannels, defaultMsgSizes)
 		node, err := newnode(*cfg, mcfg, chainID, reactor)
 		require.NoError(t, err)
 
@@ -84,14 +84,14 @@ func TestMultipleConnections(t *testing.T) {
 	}
 
 	for _, size := range []int64{
-		500,
-		1_000,
-		2_000,
-		5_000,
-		10_000,
-		50_000,
-		100_000,
-		500_000,
+		//500,
+		//1_000,
+		//2_000,
+		//5_000,
+		//10_000,
+		//50_000,
+		//100_000,
+		//500_000,
 		1_000_000,
 		10_000_000,
 		100_000_000,

p2p/load/transport_bench_test.go

@@ -36,11 +36,11 @@ func TestTransportBench(t *testing.T) {
 	mcfg.SendRate = 50000
 	mcfg.RecvRate = 50000
 
-	reactor1 := NewMockReactor(defaultTestChannels, defaultMsgSizes)
+	reactor1 := NewMockReactor(DefaultTestChannels, defaultMsgSizes)
 	node1, err := newnode(*cfg, mcfg, "test", reactor1)
 	require.NoError(t, err)
 
-	reactor2 := NewMockReactor(defaultTestChannels, defaultMsgSizes)
+	reactor2 := NewMockReactor(DefaultTestChannels, defaultMsgSizes)
 	node2, err := newnode(*cfg, mcfg, "test", reactor2)
 	require.NoError(t, err)