diff --git a/blockchain/v0/reactor_test.go b/blockchain/v0/reactor_test.go index e03745c1a..a92d6fad1 100644 --- a/blockchain/v0/reactor_test.go +++ b/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), diff --git a/consensus/reactor_test.go b/consensus/reactor_test.go index c0346d903..a53561cff 100644 --- a/consensus/reactor_test.go +++ b/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), diff --git a/evidence/reactor_test.go b/evidence/reactor_test.go index 907e11294..2170d7589 100644 --- a/evidence/reactor_test.go +++ b/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), diff --git a/mempool/reactor_test.go b/mempool/reactor_test.go index c9eb4fba3..51ac2d79d 100644 --- a/mempool/reactor_test.go +++ b/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), diff --git a/p2p/p2ptest/network.go b/p2p/p2ptest/network.go index 98fad603b..6493933ca 100644 --- a/p2p/p2ptest/network.go +++ b/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 } diff --git a/p2p/router_test.go b/p2p/router_test.go index f809af237..b4db2ca08 100644 --- a/p2p/router_test.go +++ b/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() diff --git a/p2p/transport_memory.go b/p2p/transport_memory.go index 5dd30f257..f1f04ac85 100644 --- a/p2p/transport_memory.go +++ b/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) diff --git a/p2p/transport_memory_test.go b/p2p/transport_memory_test.go index 31f896d19..476248184 100644 --- a/p2p/transport_memory_test.go +++ b/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)))