test infra: make memory network test buffers configurable (#6275)

blockchain/v0/reactor_test.go

@@ -51,7 +51,7 @@ func setup(
 
 	rts := &reactorTestSuite{
 		logger:             log.TestingLogger().With("module", "blockchain", "testCase", t.Name()),
-		network:            p2ptest.MakeNetwork(t, numNodes),
+		network:            p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: numNodes}),
 		nodes:              make([]p2p.NodeID, 0, numNodes),
 		reactors:           make(map[p2p.NodeID]*Reactor, numNodes),
 		app:                make(map[p2p.NodeID]proxy.AppConns, numNodes),

consensus/reactor_test.go

@@ -50,7 +50,7 @@ func setup(t *testing.T, numNodes int, states []*State, size int) *reactorTestSu
 	t.Helper()
 
 	rts := &reactorTestSuite{
-		network:  p2ptest.MakeNetwork(t, numNodes),
+		network:  p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: numNodes}),
 		states:   make(map[p2p.NodeID]*State),
 		reactors: make(map[p2p.NodeID]*Reactor, numNodes),
 		subs:     make(map[p2p.NodeID]types.Subscription, numNodes),

evidence/reactor_test.go

@@ -55,7 +55,7 @@ func setup(t *testing.T, stateStores []sm.Store, chBuf uint) *reactorTestSuite {
 	rts := &reactorTestSuite{
 		numStateStores: numStateStores,
 		logger:         log.TestingLogger().With("testCase", t.Name()),
-		network:        p2ptest.MakeNetwork(t, numStateStores),
+		network:        p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: numStateStores}),
 		reactors:       make(map[p2p.NodeID]*evidence.Reactor, numStateStores),
 		pools:          make(map[p2p.NodeID]*evidence.Pool, numStateStores),
 		peerUpdates:    make(map[p2p.NodeID]*p2p.PeerUpdates, numStateStores),

mempool/reactor_test.go

@@ -39,7 +39,7 @@ func setup(t *testing.T, cfg *cfg.MempoolConfig, numNodes int, chBuf uint) *reac
 
 	rts := &reactorTestSuite{
 		logger:         log.TestingLogger().With("testCase", t.Name()),
-		network:        p2ptest.MakeNetwork(t, numNodes),
+		network:        p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: numNodes}),
 		reactors:       make(map[p2p.NodeID]*Reactor, numNodes),
 		mempoolChnnels: make(map[p2p.NodeID]*p2p.Channel, numNodes),
 		mempools:       make(map[p2p.NodeID]*CListMempool, numNodes),

p2p/p2ptest/network.go

@@ -25,17 +25,31 @@ type Network struct {
 	memoryNetwork *p2p.MemoryNetwork
 }
 
+// NetworkOptions is an argument structure to parameterize the
+// MakeNetwork function.
+type NetworkOptions struct {
+	NumNodes   int
+	BufferSize int
+}
+
+func (opts *NetworkOptions) setDefaults() {
+	if opts.BufferSize == 0 {
+		opts.BufferSize = 1
+	}
+}
+
 // MakeNetwork creates a test network with the given number of nodes and
 // connects them to each other.
-func MakeNetwork(t *testing.T, nodes int) *Network {
+func MakeNetwork(t *testing.T, opts NetworkOptions) *Network {
+	opts.setDefaults()
 	logger := log.TestingLogger()
 	network := &Network{
 		Nodes:         map[p2p.NodeID]*Node{},
 		logger:        logger,
-		memoryNetwork: p2p.NewMemoryNetwork(logger),
+		memoryNetwork: p2p.NewMemoryNetwork(logger, opts.BufferSize),
 	}
 
-	for i := 0; i < nodes; i++ {
+	for i := 0; i < opts.NumNodes; i++ {
 		node := network.MakeNode(t)
 		network.Nodes[node.NodeID] = node
 	}

p2p/router_test.go

@@ -44,7 +44,7 @@ func TestRouter_Network(t *testing.T) {
 	t.Cleanup(leaktest.Check(t))
 
 	// Create a test network and open a channel where all peers run echoReactor.
-	network := p2ptest.MakeNetwork(t, 8)
+	network := p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: 8})
 	network.Start(t)
 
 	local := network.RandomNode()
@@ -143,7 +143,7 @@ func TestRouter_Channel_SendReceive(t *testing.T) {
 	t.Cleanup(leaktest.Check(t))
 
 	// Create a test network and open a channel on all nodes.
-	network := p2ptest.MakeNetwork(t, 3)
+	network := p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: 3})
 	network.Start(t)
 
 	ids := network.NodeIDs()
@@ -201,7 +201,7 @@ func TestRouter_Channel_Broadcast(t *testing.T) {
 	t.Cleanup(leaktest.Check(t))
 
 	// Create a test network and open a channel on all nodes.
-	network := p2ptest.MakeNetwork(t, 4)
+	network := p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: 4})
 	network.Start(t)
 
 	ids := network.NodeIDs()
@@ -228,7 +228,7 @@ func TestRouter_Channel_Wrapper(t *testing.T) {
 	t.Cleanup(leaktest.Check(t))
 
 	// Create a test network and open a channel on all nodes.
-	network := p2ptest.MakeNetwork(t, 2)
+	network := p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: 2})
 	network.Start(t)
 
 	ids := network.NodeIDs()
@@ -286,7 +286,7 @@ func TestRouter_Channel_Error(t *testing.T) {
 	t.Cleanup(leaktest.Check(t))
 
 	// Create a test network and open a channel on all nodes.
-	network := p2ptest.MakeNetwork(t, 3)
+	network := p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: 3})
 	network.Start(t)
 
 	ids := network.NodeIDs()

p2p/transport_memory.go

@@ -15,9 +15,6 @@ import (
 
 const (
 	MemoryProtocol Protocol = "memory"
-
-	// bufferSize is the channel buffer size of MemoryConnection.
-	bufferSize = 1
 )
 
 // MemoryNetwork is an in-memory "network" that uses buffered Go channels to
@@ -30,11 +27,13 @@ type MemoryNetwork struct {
 
 	mtx        sync.RWMutex
 	transports map[NodeID]*MemoryTransport
+	bufferSize int
 }
 
 // NewMemoryNetwork creates a new in-memory network.
-func NewMemoryNetwork(logger log.Logger) *MemoryNetwork {
+func NewMemoryNetwork(logger log.Logger, bufferSize int) *MemoryNetwork {
 	return &MemoryNetwork{
+		bufferSize: bufferSize,
 		logger:     logger,
 		transports: map[NodeID]*MemoryTransport{},
 	}
@@ -89,9 +88,10 @@ func (n *MemoryNetwork) Size() int {
 // New transports are allocated with MemoryNetwork.CreateTransport(). To contact
 // a different endpoint, both transports must be in the same MemoryNetwork.
 type MemoryTransport struct {
-	logger  log.Logger
-	network *MemoryNetwork
-	nodeID  NodeID
+	logger     log.Logger
+	network    *MemoryNetwork
+	nodeID     NodeID
+	bufferSize int
 
 	acceptCh  chan *MemoryConnection
 	closeCh   chan struct{}
@@ -102,12 +102,12 @@ type MemoryTransport struct {
 // MemoryNetwork, use MemoryNetwork.CreateTransport() instead.
 func newMemoryTransport(network *MemoryNetwork, nodeID NodeID) *MemoryTransport {
 	return &MemoryTransport{
-		logger:  network.logger.With("local", nodeID),
-		network: network,
-		nodeID:  nodeID,
-
-		acceptCh: make(chan *MemoryConnection),
-		closeCh:  make(chan struct{}),
+		logger:     network.logger.With("local", nodeID),
+		network:    network,
+		nodeID:     nodeID,
+		bufferSize: network.bufferSize,
+		acceptCh:   make(chan *MemoryConnection),
+		closeCh:    make(chan struct{}),
 	}
 }
 
@@ -164,8 +164,8 @@ func (t *MemoryTransport) Dial(ctx context.Context, endpoint Endpoint) (Connecti
 		return nil, fmt.Errorf("unknown peer %q", nodeID)
 	}
 
-	inCh := make(chan memoryMessage, bufferSize)
-	outCh := make(chan memoryMessage, bufferSize)
+	inCh := make(chan memoryMessage, t.bufferSize)
+	outCh := make(chan memoryMessage, t.bufferSize)
 	closer := tmsync.NewCloser()
 
 	outConn := newMemoryConnection(t.logger, t.nodeID, peer.nodeID, inCh, outCh, closer)

p2p/transport_memory_test.go

@@ -17,7 +17,7 @@ func init() {
 
 	testTransports["memory"] = func(t *testing.T) p2p.Transport {
 		if network == nil {
-			network = p2p.NewMemoryNetwork(log.TestingLogger())
+			network = p2p.NewMemoryNetwork(log.TestingLogger(), 1)
 		}
 		i := byte(network.Size())
 		nodeID, err := p2p.NewNodeID(hex.EncodeToString(bytes.Repeat([]byte{i<<4 + i}, 20)))