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

dot/network/service.go

@@ -174,6 +174,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 {
@@ -503,8 +504,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 {
@@ -691,13 +690,11 @@ func (s *Service) ReportPeer(change peerset.ReputationChange, p peer.ID) {
 }
 
 func (s *Service) startPeerSetHandler() {
-	s.host.cm.peerSetHandler.Start(s.ctx)
+	s.host.cm.peerSetHandler.Start(s.ctx, s.processMessage)
 	// wait for peerSetHandler to start.
 	if !s.noBootstrap {
 		s.host.bootstrap()
 	}
-
-	go s.startProcessingMsg()
 }
 
 func (s *Service) processMessage(msg peerset.Message) {
@@ -733,19 +730,3 @@ func (s *Service) processMessage(msg peerset.Message) {
 		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

@@ -47,8 +47,7 @@ type TransactionHandler interface {
 
 // PeerSetHandler is the interface used by the connection manager to handle peerset.
 type PeerSetHandler interface {
-	Start(context.Context)
-	Stop()
+	Start(context.Context, func(peerset.Message))
 	ReportPeer(peerset.ReputationChange, ...peer.ID)
 	PeerAdd
 	PeerRemove
@@ -73,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 %v, 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, nil, stringfyPeers(peers))
+		logger.Errorf("failed to do action %s on peerSet: %s", msg, err)
 	}
 }
 
 // RemoveReservedPeer remove 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, nil, stringfyPeers(peers))
+		logger.Errorf("failed to do action %s on peerSet: %s", msg, err)
 	}
 }
 
 // SetReservedPeer set 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, might required to implement RPC Call for this.
+	err := h.peerSet.setReservedPeer(setID, peers...)
+	if err != nil {
+		msg := fmt.Sprintf(logStringPattern, setReservedPeers, setID, nil, 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, nil, 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, nil, 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, 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, nil, 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, nil, stringfyPeers(peers))
+		logger.Errorf("failed to do action %s on peerSet: %s", msg, err)
 	}
 }
 
@@ -113,40 +115,24 @@ func (h *Handler) PeerReputation(peerID peer.ID) (Reputation, error) {
 	if err != nil {
 		return 0, err
 	}
-	return n.getReputation(), nil
+	return n.rep, nil
 }
 
 // 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)
+func (h *Handler) Start(ctx context.Context, processMessageFn func(Message)) {
+	h.peerSet.start(ctx, processMessageFn)
 }
 
 // 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
+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, ", ")
 }