Merge pull request #1702 from tendermint/xla/fix-racy-config

Fix race when mutating MConnConfig

config/config.go

@@ -5,8 +5,6 @@ import (
 	"os"
 	"path/filepath"
 	"time"
-
-	tmconn "github.com/tendermint/tendermint/p2p/conn"
 )
 
 const (
@@ -304,9 +302,8 @@ type P2PConfig struct {
 	AllowDuplicateIP bool `mapstructure:"allow_duplicate_ip"`
 
 	// Peer connection configuration.
-	HandshakeTimeout time.Duration      `mapstructure:"handshake_timeout"`
-	DialTimeout      time.Duration      `mapstructure:"dial_timeout"`
-	MConfig          tmconn.MConnConfig `mapstructure:"connection"`
+	HandshakeTimeout time.Duration `mapstructure:"handshake_timeout"`
+	DialTimeout      time.Duration `mapstructure:"dial_timeout"`
 
 	// Testing params.
 	// Force dial to fail
@@ -332,7 +329,6 @@ func DefaultP2PConfig() *P2PConfig {
 		AllowDuplicateIP:        true, // so non-breaking yet
 		HandshakeTimeout:        20 * time.Second,
 		DialTimeout:             3 * time.Second,
-		MConfig:                 tmconn.DefaultMConnConfig(),
 		TestDialFail:            false,
 		TestFuzz:                false,
 		TestFuzzConfig:          DefaultFuzzConnConfig(),

p2p/peer.go

@@ -102,6 +102,7 @@ type peer struct {
 
 func newPeer(
 	pc peerConn,
+	mConfig tmconn.MConnConfig,
 	nodeInfo NodeInfo,
 	reactorsByCh map[byte]Reactor,
 	chDescs []*tmconn.ChannelDescriptor,
@@ -120,7 +121,7 @@ func newPeer(
 		reactorsByCh,
 		chDescs,
 		onPeerError,
-		pc.config.MConfig,
+		mConfig,
 	)
 	p.BaseService = *cmn.NewBaseService(nil, "Peer", p)
 

p2p/peer_test.go

@@ -27,7 +27,7 @@ func TestPeerBasic(t *testing.T) {
 	rp.Start()
 	defer rp.Stop()
 
-	p, err := createOutboundPeerAndPerformHandshake(rp.Addr(), cfg)
+	p, err := createOutboundPeerAndPerformHandshake(rp.Addr(), cfg, tmconn.DefaultMConnConfig())
 	require.Nil(err)
 
 	err = p.Start()
@@ -53,7 +53,7 @@ func TestPeerSend(t *testing.T) {
 	rp.Start()
 	defer rp.Stop()
 
-	p, err := createOutboundPeerAndPerformHandshake(rp.Addr(), config)
+	p, err := createOutboundPeerAndPerformHandshake(rp.Addr(), config, tmconn.DefaultMConnConfig())
 	require.Nil(err)
 
 	err = p.Start()
@@ -65,7 +65,11 @@ func TestPeerSend(t *testing.T) {
 	assert.True(p.Send(testCh, []byte("Asylum")))
 }
 
-func createOutboundPeerAndPerformHandshake(addr *NetAddress, config *config.P2PConfig) (*peer, error) {
+func createOutboundPeerAndPerformHandshake(
+	addr *NetAddress,
+	config *config.P2PConfig,
+	mConfig tmconn.MConnConfig,
+) (*peer, error) {
 	chDescs := []*tmconn.ChannelDescriptor{
 		{ID: testCh, Priority: 1},
 	}
@@ -86,7 +90,7 @@ func createOutboundPeerAndPerformHandshake(addr *NetAddress, config *config.P2PC
 		return nil, err
 	}
 
-	p := newPeer(pc, nodeInfo, reactorsByCh, chDescs, func(p Peer, r interface{}) {})
+	p := newPeer(pc, mConfig, nodeInfo, reactorsByCh, chDescs, func(p Peer, r interface{}) {})
 	p.SetLogger(log.TestingLogger().With("peer", addr))
 	return p, nil
 }

p2p/switch.go

@@ -70,6 +70,8 @@ type Switch struct {
 	filterConnByAddr func(net.Addr) error
 	filterConnByID   func(ID) error
 
+	mConfig conn.MConnConfig
+
 	rng *cmn.Rand // seed for randomizing dial times and orders
 }
 
@@ -88,10 +90,13 @@ func NewSwitch(cfg *config.P2PConfig) *Switch {
 	// Ensure we have a completely undeterministic PRNG.
 	sw.rng = cmn.NewRand()
 
-	sw.config.MConfig.FlushThrottle = time.Duration(cfg.FlushThrottleTimeout) * time.Millisecond
-	sw.config.MConfig.SendRate = cfg.SendRate
-	sw.config.MConfig.RecvRate = cfg.RecvRate
-	sw.config.MConfig.MaxPacketMsgPayloadSize = cfg.MaxPacketMsgPayloadSize
+	mConfig := conn.DefaultMConnConfig()
+	mConfig.FlushThrottle = time.Duration(cfg.FlushThrottleTimeout) * time.Millisecond
+	mConfig.SendRate = cfg.SendRate
+	mConfig.RecvRate = cfg.RecvRate
+	mConfig.MaxPacketMsgPayloadSize = cfg.MaxPacketMsgPayloadSize
+
+	sw.mConfig = mConfig
 
 	sw.BaseService = *cmn.NewBaseService(nil, "P2P Switch", sw)
 	return sw
@@ -600,7 +605,7 @@ func (sw *Switch) addPeer(pc peerConn) error {
 		return err
 	}
 
-	peer := newPeer(pc, peerNodeInfo, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError)
+	peer := newPeer(pc, sw.mConfig, peerNodeInfo, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError)
 	peer.SetLogger(sw.Logger.With("peer", addr))
 
 	peer.Logger.Info("Successful handshake with peer", "peerNodeInfo", peerNodeInfo)