fix(dot/peerset): remove race conditions from peerset package

dot/network/host.go

@@ -223,6 +223,7 @@ func newHost(ctx context.Context, cfg *Config) (*host, error) {
 	}
 
 	cm.host = host
+	cm.peerSetHandler.SetMessageProcessor(host)
 	return host, nil
 }
 
@@ -437,3 +438,38 @@ func (h *host) closeProtocolStream(pID protocol.ID, p peer.ID) {
 		}
 	}
 }
+
+// Process will connect, drop or reject a peer based on a peerset message
+func (h *host) Process(msg peerset.Message) {
+	peerID := msg.PeerID
+	if peerID == "" {
+		logger.Errorf("found empty peer id in peerset message")
+		return
+	}
+	switch msg.Status {
+	case peerset.Connect:
+		addrInfo := h.h.Peerstore().PeerInfo(peerID)
+		if len(addrInfo.Addrs) == 0 {
+			var err error
+			addrInfo, err = h.discovery.findPeer(peerID)
+			if err != nil {
+				logger.Warnf("failed to find peer id %s: %s", peerID, err)
+				return
+			}
+		}
+
+		err := h.connect(addrInfo)
+		if err != nil {
+			logger.Warnf("failed to open connection for peer %s: %s", peerID, err)
+			return
+		}
+		logger.Debugf("connection successful with peer %s", peerID)
+	case peerset.Drop, peerset.Reject:
+		err := h.closePeer(peerID)
+		if err != nil {
+			logger.Warnf("failed to close connection with peer %s: %s", peerID, err)
+			return
+		}
+		logger.Debugf("connection dropped successfully for peer %s", peerID)
+	}
+}

dot/network/service.go

@@ -135,8 +135,7 @@ type Service struct {
 
 	blockResponseBuf   []byte
 	blockResponseBufMu sync.Mutex
-
-	telemetry telemetry.Client
+	telemetry          telemetry.Client
 }
 
 // NewService creates a new network service from the configuration and message channels
@@ -174,6 +173,7 @@ func NewService(cfg *Config) (*Service, error) {
 	if cfg.batchSize == 0 {
 		cfg.batchSize = defaultTxnBatchSize
 	}
+
 	// create a new host instance
 	host, err := newHost(ctx, cfg)
 	if err != nil {
@@ -487,8 +487,6 @@ func (s *Service) Stop() error {
 		logger.Errorf("Failed to close host: %s", err)
 	}
 
-	s.host.cm.peerSetHandler.Stop()
-
 	// check if closeCh is closed, if not, close it.
 mainloop:
 	for {
@@ -680,56 +678,4 @@ func (s *Service) startPeerSetHandler() {
 	if !s.noBootstrap {
 		s.host.bootstrap()
 	}
-
-	go s.startProcessingMsg()
-}
-
-func (s *Service) processMessage(msg peerset.Message) {
-	peerID := msg.PeerID
-	if peerID == "" {
-		logger.Errorf("found empty peer id in peerset message")
-		return
-	}
-	switch msg.Status {
-	case peerset.Connect:
-		addrInfo := s.host.h.Peerstore().PeerInfo(peerID)
-		if len(addrInfo.Addrs) == 0 {
-			var err error
-			addrInfo, err = s.host.discovery.findPeer(peerID)
-			if err != nil {
-				logger.Warnf("failed to find peer id %s: %s", peerID, err)
-				return
-			}
-		}
-
-		err := s.host.connect(addrInfo)
-		if err != nil {
-			logger.Warnf("failed to open connection for peer %s: %s", peerID, err)
-			return
-		}
-		logger.Debugf("connection successful with peer %s", peerID)
-	case peerset.Drop, peerset.Reject:
-		err := s.host.closePeer(peerID)
-		if err != nil {
-			logger.Warnf("failed to close connection with peer %s: %s", peerID, err)
-			return
-		}
-		logger.Debugf("connection dropped successfully for peer %s", peerID)
-	}
-}
-
-func (s *Service) startProcessingMsg() {
-	msgCh := s.host.cm.peerSetHandler.Messages()
-	for {
-		select {
-		case <-s.ctx.Done():
-			return
-		case msg, ok := <-msgCh:
-			if !ok {
-				return
-			}
-
-			s.processMessage(msg)
-		}
-	}
 }

dot/network/state.go

@@ -46,8 +46,8 @@ type TransactionHandler interface {
 
 // PeerSetHandler is the interface used by the connection manager to handle peerset.
 type PeerSetHandler interface {
+	SetMessageProcessor(peerset.MessageProcessor)
 	Start(context.Context)
-	Stop()
 	ReportPeer(peerset.ReputationChange, ...peer.ID)
 	PeerAdd
 	PeerRemove
@@ -72,6 +72,4 @@ type PeerRemove interface {
 // Peer is the interface used by the PeerSetHandler to get the peer data from peerSet.
 type Peer interface {
 	PeerReputation(peer.ID) (peerset.Reputation, error)
-	SortedPeers(idx int) chan peer.IDSlice
-	Messages() chan peerset.Message
 }

dot/peerset/handler.go

@@ -5,17 +5,17 @@ package peerset
 
 import (
 	"context"
+	"fmt"
+	"strings"
 
 	"github.com/libp2p/go-libp2p-core/peer"
 )
 
+const logStringPattern = "call=%s, set-id=%d, reputation change %s, peers=[%s]"
+
 // Handler manages peerSet.
 type Handler struct {
-	actionQueue chan<- action
-	peerSet     *PeerSet
-	closeCh     chan struct{}
-
-	cancelCtx context.CancelFunc
+	peerSet *PeerSet
 }
 
 // NewPeerSetHandler creates a new *peerset.Handler.
@@ -30,80 +30,82 @@ func NewPeerSetHandler(cfg *ConfigSet) (*Handler, error) {
 	}, nil
 }
 
+// SetReservedOnlyPeer not yet implemented
+func (h *Handler) SetReservedOnlyPeer(setID int, peers ...peer.ID) {
+	// TODO: not yet implemented (#1888)
+	logger.Errorf("failed to do action %s on peerSet: not implemented yet", setReservedOnly)
+}
+
 // AddReservedPeer adds reserved peer into peerSet.
 func (h *Handler) AddReservedPeer(setID int, peers ...peer.ID) {
-	h.actionQueue <- action{
-		actionCall: addReservedPeer,
-		setID:      setID,
-		peers:      peers,
+	err := h.peerSet.addReservedPeers(setID, peers...)
+	if err != nil {
+		msg := fmt.Sprintf(logStringPattern, addReservedPeer, setID, "", stringfyPeers(peers))
+		logger.Errorf("failed to do action %s on peerSet: %s", msg, err)
 	}
 }
 
-// RemoveReservedPeer remove reserved peer from peerSet.
+// RemoveReservedPeer removes reserved peer from peerSet.
 func (h *Handler) RemoveReservedPeer(setID int, peers ...peer.ID) {
-	h.actionQueue <- action{
-		actionCall: removeReservedPeer,
-		setID:      setID,
-		peers:      peers,
+	err := h.peerSet.removeReservedPeers(setID, peers...)
+	if err != nil {
+		msg := fmt.Sprintf(logStringPattern, removeReservedPeer, setID, "", stringfyPeers(peers))
+		logger.Errorf("failed to do action %s on peerSet: %s", msg, err)
 	}
 }
 
-// SetReservedPeer set the reserve peer into peerSet
+// SetReservedPeer sets the reserve peer into peerSet
 func (h *Handler) SetReservedPeer(setID int, peers ...peer.ID) {
-	h.actionQueue <- action{
-		actionCall: setReservedPeers,
-		setID:      setID,
-		peers:      peers,
+	// TODO: this is not used yet, it might be required to implement an RPC Call for this.
+	err := h.peerSet.setReservedPeer(setID, peers...)
+	if err != nil {
+		msg := fmt.Sprintf(logStringPattern, setReservedPeers, setID, "", stringfyPeers(peers))
+		logger.Errorf("failed to do action %s on peerSet: %s", msg, err)
 	}
 }
 
 // AddPeer adds peer to peerSet.
 func (h *Handler) AddPeer(setID int, peers ...peer.ID) {
-	h.actionQueue <- action{
-		actionCall: addToPeerSet,
-		setID:      setID,
-		peers:      peers,
+	err := h.peerSet.addPeer(setID, peers)
+	if err != nil {
+		msg := fmt.Sprintf(logStringPattern, addToPeerSet, setID, "", stringfyPeers(peers))
+		logger.Errorf("failed to do action %s on peerSet: %s", msg, err)
 	}
 }
 
 // RemovePeer removes peer from peerSet.
 func (h *Handler) RemovePeer(setID int, peers ...peer.ID) {
-	h.actionQueue <- action{
-		actionCall: removeFromPeerSet,
-		setID:      setID,
-		peers:      peers,
+	err := h.peerSet.removePeer(setID, peers...)
+	if err != nil {
+		msg := fmt.Sprintf(logStringPattern, removeFromPeerSet, setID, "", stringfyPeers(peers))
+		logger.Errorf("failed to do action %s on peerSet: %s", msg, err)
 	}
 }
 
 // ReportPeer reports ReputationChange according to the peer behaviour.
 func (h *Handler) ReportPeer(rep ReputationChange, peers ...peer.ID) {
-	h.actionQueue <- action{
-		actionCall: reportPeer,
-		reputation: rep,
-		peers:      peers,
+	err := h.peerSet.reportPeer(rep, peers...)
+	if err != nil {
+		msg := fmt.Sprintf(logStringPattern, reportPeer, 0, rep.String(), stringfyPeers(peers))
+		logger.Errorf("failed to do action %s on peerSet: %s", msg, err)
 	}
 }
 
 // Incoming calls when we have an incoming connection from peer.
 func (h *Handler) Incoming(setID int, peers ...peer.ID) {
-	h.actionQueue <- action{
-		actionCall: incoming,
-		peers:      peers,
-		setID:      setID,
+	err := h.peerSet.incoming(setID, peers...)
+	if err != nil {
+		msg := fmt.Sprintf(logStringPattern, incoming, setID, "", stringfyPeers(peers))
+		logger.Errorf("failed to do action %s on peerSet: %s", msg, err)
 	}
 }
 
-// Messages return result message chan.
-func (h *Handler) Messages() chan Message {
-	return h.peerSet.resultMsgCh
-}
-
 // DisconnectPeer calls for disconnecting a connection from peer.
 func (h *Handler) DisconnectPeer(setID int, peers ...peer.ID) {
-	h.actionQueue <- action{
-		actionCall: disconnect,
-		setID:      setID,
-		peers:      peers,
+	err := h.peerSet.disconnect(setID, UnknownDrop, peers...)
+	if err != nil {
+		msg := fmt.Sprintf(logStringPattern, disconnect, setID, "", stringfyPeers(peers))
+		logger.Errorf("failed to do action %s on peerSet: %s", msg, err)
 	}
 }
 
@@ -113,40 +115,29 @@ func (h *Handler) PeerReputation(peerID peer.ID) (Reputation, error) {
 	if err != nil {
 		return 0, err
 	}
-	return n.getReputation(), nil
+	return n.reputation, nil
+}
+
+// SetMessageProcessor set a processor to process peerset messages
+func (h *Handler) SetMessageProcessor(processor MessageProcessor) {
+	h.peerSet.processor = processor
 }
 
 // Start starts peerSet processing
 func (h *Handler) Start(ctx context.Context) {
-	ctx, cancel := context.WithCancel(ctx)
-	h.cancelCtx = cancel
-
-	actionCh := make(chan action, msgChanSize)
-	h.closeCh = make(chan struct{})
-	h.actionQueue = actionCh
-
-	h.peerSet.start(ctx, actionCh)
+	h.peerSet.start(ctx)
 }
 
-// SortedPeers return chan for sorted connected peer in the peerSet.
-func (h *Handler) SortedPeers(setIdx int) chan peer.IDSlice {
-	resultPeersCh := make(chan peer.IDSlice)
-	h.actionQueue <- action{
-		actionCall:    sortedPeers,
-		resultPeersCh: resultPeersCh,
-		setID:         setIdx,
-	}
-
-	return resultPeersCh
+// SortedPeers returns a sorted peer ID slice for connected peers in the peerSet.
+func (h *Handler) SortedPeers(setIdx int) peer.IDSlice {
+	return h.peerSet.peerState.sortedPeers(setIdx)
 }
 
-// Stop closes the actionQueue and result message chan.
-func (h *Handler) Stop() {
-	select {
-	case <-h.closeCh:
-	default:
-		h.cancelCtx()
-		close(h.closeCh)
-		close(h.actionQueue)
+func stringfyPeers(peers peer.IDSlice) string {
+	peersStrings := make([]string, len(peers))
+	for i := range peers {
+		peersStrings[i] = peers[i].String()
 	}
+
+	return strings.Join(peersStrings, ", ")
 }