Merge pull request #5 from sei-protocol/tony-chen-fix-connections

Fix open connection race condition

Author: codchen

internal/blocksync/reactor.go

@@ -11,7 +11,6 @@ import (
 	"github.com/tendermint/tendermint/internal/consensus"
 	"github.com/tendermint/tendermint/internal/eventbus"
 	"github.com/tendermint/tendermint/internal/p2p"
-	"github.com/tendermint/tendermint/internal/p2p/conn"
 	sm "github.com/tendermint/tendermint/internal/state"
 	"github.com/tendermint/tendermint/internal/store"
 	"github.com/tendermint/tendermint/libs/log"
@@ -81,8 +80,8 @@ type Reactor struct {
 	consReactor consensusReactor
 	blockSync   *atomicBool
 
-	chCreator  p2p.ChannelCreator
 	peerEvents p2p.PeerEventSubscriber
+	channel    *p2p.Channel
 
 	requestsCh <-chan BlockRequest
 	errorsCh   <-chan peerError
@@ -100,7 +99,6 @@ func NewReactor(
 	blockExec *sm.BlockExecutor,
 	store *store.BlockStore,
 	consReactor consensusReactor,
-	channelCreator p2p.ChannelCreator,
 	peerEvents p2p.PeerEventSubscriber,
 	blockSync bool,
 	metrics *consensus.Metrics,
@@ -113,7 +111,6 @@ func NewReactor(
 		store:       store,
 		consReactor: consReactor,
 		blockSync:   newAtomicBool(blockSync),
-		chCreator:   channelCreator,
 		peerEvents:  peerEvents,
 		metrics:     metrics,
 		eventBus:    eventBus,
@@ -123,6 +120,10 @@ func NewReactor(
 	return r
 }
 
+func (r *Reactor) SetChannel(ch *p2p.Channel) {
+	r.channel = ch
+}
+
 // OnStart starts separate go routines for each p2p Channel and listens for
 // envelopes on each. In addition, it also listens for peer updates and handles
 // messages on that p2p channel accordingly. The caller must be sure to execute
@@ -131,12 +132,6 @@ func NewReactor(
 // If blockSync is enabled, we also start the pool and the pool processing
 // goroutine. If the pool fails to start, an error is returned.
 func (r *Reactor) OnStart(ctx context.Context) error {
-	blockSyncCh, err := r.chCreator(ctx, GetChannelDescriptor())
-	if err != nil {
-		return err
-	}
-	r.chCreator = func(context.Context, *conn.ChannelDescriptor) (*p2p.Channel, error) { return blockSyncCh, nil }
-
 	state, err := r.stateStore.Load()
 	if err != nil {
 		return err
@@ -162,13 +157,13 @@ func (r *Reactor) OnStart(ctx context.Context) error {
 		if err := r.pool.Start(ctx); err != nil {
 			return err
 		}
-		go r.requestRoutine(ctx, blockSyncCh)
+		go r.requestRoutine(ctx, r.channel)
 
-		go r.poolRoutine(ctx, false, blockSyncCh)
+		go r.poolRoutine(ctx, false, r.channel)
 	}
 
-	go r.processBlockSyncCh(ctx, blockSyncCh)
-	go r.processPeerUpdates(ctx, r.peerEvents(ctx), blockSyncCh)
+	go r.processBlockSyncCh(ctx, r.channel)
+	go r.processPeerUpdates(ctx, r.peerEvents(ctx), r.channel)
 
 	return nil
 }
@@ -378,13 +373,8 @@ func (r *Reactor) SwitchToBlockSync(ctx context.Context, state sm.State) error {
 
 	r.syncStartTime = time.Now()
 
-	bsCh, err := r.chCreator(ctx, GetChannelDescriptor())
-	if err != nil {
-		return err
-	}
-
-	go r.requestRoutine(ctx, bsCh)
-	go r.poolRoutine(ctx, true, bsCh)
+	go r.requestRoutine(ctx, r.channel)
+	go r.poolRoutine(ctx, true, r.channel)
 
 	if err := r.PublishStatus(types.EventDataBlockSyncStatus{
 		Complete: false,

internal/blocksync/reactor_test.go

@@ -149,7 +149,6 @@ func makeReactor(
 		blockExec,
 		blockStore,
 		nil,
-		channelCreator,
 		peerEvents,
 		true,
 		consensus.NopMetrics(),

internal/consensus/reactor.go

@@ -27,49 +27,54 @@ var (
 	_ p2p.Wrapper     = (*tmcons.Message)(nil)
 )
 
-// GetChannelDescriptor produces an instance of a descriptor for this
-// package's required channels.
-func getChannelDescriptors() map[p2p.ChannelID]*p2p.ChannelDescriptor {
-	return map[p2p.ChannelID]*p2p.ChannelDescriptor{
-		StateChannel: {
-			ID:                  StateChannel,
-			MessageType:         new(tmcons.Message),
-			Priority:            8,
-			SendQueueCapacity:   64,
-			RecvMessageCapacity: maxMsgSize,
-			RecvBufferCapacity:  128,
-			Name:                "state",
-		},
-		DataChannel: {
-			// TODO: Consider a split between gossiping current block and catchup
-			// stuff. Once we gossip the whole block there is nothing left to send
-			// until next height or round.
-			ID:                  DataChannel,
-			MessageType:         new(tmcons.Message),
-			Priority:            12,
-			SendQueueCapacity:   64,
-			RecvBufferCapacity:  512,
-			RecvMessageCapacity: maxMsgSize,
-			Name:                "data",
-		},
-		VoteChannel: {
-			ID:                  VoteChannel,
-			MessageType:         new(tmcons.Message),
-			Priority:            10,
-			SendQueueCapacity:   64,
-			RecvBufferCapacity:  128,
-			RecvMessageCapacity: maxMsgSize,
-			Name:                "vote",
-		},
-		VoteSetBitsChannel: {
-			ID:                  VoteSetBitsChannel,
-			MessageType:         new(tmcons.Message),
-			Priority:            5,
-			SendQueueCapacity:   8,
-			RecvBufferCapacity:  128,
-			RecvMessageCapacity: maxMsgSize,
-			Name:                "voteSet",
-		},
+func GetStateChannelDescriptor() *p2p.ChannelDescriptor {
+	return &p2p.ChannelDescriptor{
+		ID:                  StateChannel,
+		MessageType:         new(tmcons.Message),
+		Priority:            8,
+		SendQueueCapacity:   64,
+		RecvMessageCapacity: maxMsgSize,
+		RecvBufferCapacity:  128,
+		Name:                "state",
+	}
+}
+
+func GetDataChannelDescriptor() *p2p.ChannelDescriptor {
+	return &p2p.ChannelDescriptor{
+		// TODO: Consider a split between gossiping current block and catchup
+		// stuff. Once we gossip the whole block there is nothing left to send
+		// until next height or round.
+		ID:                  DataChannel,
+		MessageType:         new(tmcons.Message),
+		Priority:            12,
+		SendQueueCapacity:   64,
+		RecvBufferCapacity:  512,
+		RecvMessageCapacity: maxMsgSize,
+		Name:                "data",
+	}
+}
+
+func GetVoteChannelDescriptor() *p2p.ChannelDescriptor {
+	return &p2p.ChannelDescriptor{
+		ID:                  VoteChannel,
+		MessageType:         new(tmcons.Message),
+		Priority:            10,
+		SendQueueCapacity:   64,
+		RecvBufferCapacity:  128,
+		RecvMessageCapacity: maxMsgSize,
+		Name:                "vote",
+	}
+}
+
+func GetVoteSetChannelDescriptor() *p2p.ChannelDescriptor {
+	return &p2p.ChannelDescriptor{
+		ID:                  VoteSetBitsChannel,
+		MessageType:         new(tmcons.Message),
+		Priority:            5,
+		SendQueueCapacity:   8,
+		RecvBufferCapacity:  128,
+		RecvMessageCapacity: maxMsgSize,
+		Name:                "voteSet",
 	}
 }
 
@@ -103,8 +108,9 @@ type BlockSyncReactor interface {
 	GetRemainingSyncTime() time.Duration
 }
 
-//go:generate ../../scripts/mockery_generate.sh ConsSyncReactor
 // ConsSyncReactor defines an interface used for testing abilities of node.startStateSync.
+//
+//go:generate ../../scripts/mockery_generate.sh ConsSyncReactor
 type ConsSyncReactor interface {
 	SwitchToConsensus(sm.State, bool)
 	SetStateSyncingMetrics(float64)
@@ -127,7 +133,8 @@ type Reactor struct {
 	readySignal chan struct{} // closed when the node is ready to start consensus
 
 	peerEvents p2p.PeerEventSubscriber
-	chCreator  p2p.ChannelCreator
+
+	channels *channelBundle
 }
 
 // NewReactor returns a reference to a new consensus reactor, which implements
@@ -137,7 +144,6 @@ type Reactor struct {
 func NewReactor(
 	logger log.Logger,
 	cs *State,
-	channelCreator p2p.ChannelCreator,
 	peerEvents p2p.PeerEventSubscriber,
 	eventBus *eventbus.EventBus,
 	waitSync bool,
@@ -152,8 +158,8 @@ func NewReactor(
 		eventBus:    eventBus,
 		Metrics:     metrics,
 		peerEvents:  peerEvents,
-		chCreator:   channelCreator,
 		readySignal: make(chan struct{}),
+		channels:    &channelBundle{},
 	}
 	r.BaseService = *service.NewBaseService(logger, "Consensus", r)
 
@@ -171,6 +177,22 @@ type channelBundle struct {
 	votSet *p2p.Channel
 }
 
+func (r *Reactor) SetStateChannel(ch *p2p.Channel) {
+	r.channels.state = ch
+}
+
+func (r *Reactor) SetDataChannel(ch *p2p.Channel) {
+	r.channels.data = ch
+}
+
+func (r *Reactor) SetVoteChannel(ch *p2p.Channel) {
+	r.channels.vote = ch
+}
+
+func (r *Reactor) SetVoteSetChannel(ch *p2p.Channel) {
+	r.channels.votSet = ch
+}
+
 // OnStart starts separate go routines for each p2p Channel and listens for
 // envelopes on each. In addition, it also listens for peer updates and handles
 // messages on that p2p channel accordingly. The caller must be sure to execute
@@ -180,37 +202,13 @@ func (r *Reactor) OnStart(ctx context.Context) error {
 
 	peerUpdates := r.peerEvents(ctx)
 
-	var chBundle channelBundle
-	var err error
-
-	chans := getChannelDescriptors()
-	chBundle.state, err = r.chCreator(ctx, chans[StateChannel])
-	if err != nil {
-		return err
-	}
-
-	chBundle.data, err = r.chCreator(ctx, chans[DataChannel])
-	if err != nil {
-		return err
-	}
-
-	chBundle.vote, err = r.chCreator(ctx, chans[VoteChannel])
-	if err != nil {
-		return err
-	}
-
-	chBundle.votSet, err = r.chCreator(ctx, chans[VoteSetBitsChannel])
-	if err != nil {
-		return err
-	}
-
 	// start routine that computes peer statistics for evaluating peer quality
 	//
 	// TODO: Evaluate if we need this to be synchronized via WaitGroup as to not
 	// leak the goroutine when stopping the reactor.
 	go r.peerStatsRoutine(ctx, peerUpdates)
 
-	r.subscribeToBroadcastEvents(ctx, chBundle.state)
+	r.subscribeToBroadcastEvents(ctx, r.channels.state)
 
 	if !r.WaitSync() {
 		if err := r.state.Start(ctx); err != nil {
@@ -222,11 +220,11 @@ func (r *Reactor) OnStart(ctx context.Context) error {
 
 	go r.updateRoundStateRoutine(ctx)
 
-	go r.processStateCh(ctx, chBundle)
-	go r.processDataCh(ctx, chBundle)
-	go r.processVoteCh(ctx, chBundle)
-	go r.processVoteSetBitsCh(ctx, chBundle)
-	go r.processPeerUpdates(ctx, peerUpdates, chBundle)
+	go r.processStateCh(ctx, *r.channels)
+	go r.processDataCh(ctx, *r.channels)
+	go r.processVoteCh(ctx, *r.channels)
+	go r.processVoteSetBitsCh(ctx, *r.channels)
+	go r.processPeerUpdates(ctx, peerUpdates, *r.channels)
 
 	return nil
 }

internal/consensus/reactor_test.go

@@ -85,31 +85,13 @@ func setup(
 	ctx, cancel := context.WithCancel(ctx)
 	t.Cleanup(cancel)
 
-	chCreator := func(nodeID types.NodeID) p2p.ChannelCreator {
-		return func(ctx context.Context, desc *p2p.ChannelDescriptor) (*p2p.Channel, error) {
-			switch desc.ID {
-			case StateChannel:
-				return rts.stateChannels[nodeID], nil
-			case DataChannel:
-				return rts.dataChannels[nodeID], nil
-			case VoteChannel:
-				return rts.voteChannels[nodeID], nil
-			case VoteSetBitsChannel:
-				return rts.voteSetBitsChannels[nodeID], nil
-			default:
-				return nil, fmt.Errorf("invalid channel; %v", desc.ID)
-			}
-		}
-	}
-
 	i := 0
 	for nodeID, node := range rts.network.Nodes {
 		state := states[i]
 
 		reactor := NewReactor(
 			state.logger.With("node", nodeID),
 			state,
-			chCreator(nodeID),
 			func(ctx context.Context) *p2p.PeerUpdates { return node.MakePeerUpdates(ctx, t) },
 			state.eventBus,
 			true,
@@ -696,7 +678,6 @@ func TestSwitchToConsensusVoteExtensions(t *testing.T) {
 				log.NewNopLogger(),
 				cs,
 				nil,
-				nil,
 				cs.eventBus,
 				true,
 				NopMetrics(),