x/sync: add locks for peerTracker

x/sync/network_client.go

@@ -178,17 +178,14 @@ func (c *networkClient) RequestAny(
 	}
 	defer c.activeRequests.Release(1)
 
-	c.lock.Lock()
 	nodeID, ok := c.peers.GetAnyPeer(minVersion)
 	if !ok {
-		c.lock.Unlock()
 		return ids.EmptyNodeID, nil, fmt.Errorf(
 			"no peers found matching version %s out of %d peers",
 			minVersion, c.peers.Size(),
 		)
 	}
 
-	// Note [c.request] releases [c.lock].
 	response, err := c.request(ctx, nodeID, request)
 	return nodeID, response, err
 }
@@ -208,8 +205,6 @@ func (c *networkClient) Request(
 	}
 	defer c.activeRequests.Release(1)
 
-	c.lock.Lock()
-	// Note [c.request] releases [c.lock].
 	return c.request(ctx, nodeID, request)
 }
 
@@ -219,12 +214,13 @@ func (c *networkClient) Request(
 // Releases active requests semaphore if there was an error in sending the request.
 // Assumes [nodeID] is never [c.myNodeID] since we guarantee
 // [c.myNodeID] will not be added to [c.peers].
-// Assumes [c.lock] is held and unlocks [c.lock] before returning.
+// Assumes [c.lock] is not held and unlocks [c.lock] before returning.
 func (c *networkClient) request(
 	ctx context.Context,
 	nodeID ids.NodeID,
 	request []byte,
 ) ([]byte, error) {
+	c.lock.Lock()
 	c.log.Debug("sending request to peer",
 		zap.Stringer("nodeID", nodeID),
 		zap.Int("requestLen", len(request)),
@@ -246,13 +242,13 @@ func (c *networkClient) request(
 	handler := newResponseHandler()
 	c.outstandingRequestHandlers[requestID] = handler
 
+	c.lock.Unlock() // unlock so response can be received
+
 	var (
 		response  []byte
 		startTime = time.Now()
 	)
 
-	c.lock.Unlock() // unlock so response can be received
-
 	select {
 	case <-ctx.Done():
 		c.peers.TrackBandwidth(nodeID, 0)
@@ -283,9 +279,6 @@ func (c *networkClient) Connected(
 	nodeID ids.NodeID,
 	nodeVersion *version.Application,
 ) error {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-
 	if nodeID == c.myNodeID {
 		c.log.Debug("skipping registering self as peer")
 		return nil
@@ -298,9 +291,6 @@ func (c *networkClient) Connected(
 
 // Disconnected removes given [nodeID] from the peer list.
 func (c *networkClient) Disconnected(_ context.Context, nodeID ids.NodeID) error {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-
 	if nodeID == c.myNodeID {
 		c.log.Debug("skipping deregistering self as peer")
 		return nil

x/sync/peer_tracker.go

@@ -5,6 +5,7 @@ package sync
 
 import (
 	"math/rand"
+	"sync"
 	"time"
 
 	stdmath "math"
@@ -40,8 +41,9 @@ type peerInfo struct {
 // Tracks the bandwidth of responses coming from peers,
 // preferring to contact peers with known good bandwidth, connecting
 // to new peers with an exponentially decaying probability.
-// Note: not thread safe. Caller must handle synchronization.
 type peerTracker struct {
+	// Lock to protect concurrent access to the peer tracker
+	lock sync.Mutex
 	// All peers we are connected to
 	peers map[ids.NodeID]*peerInfo
 	// Peers that we're connected to that we've sent a request to
@@ -75,6 +77,7 @@ func newPeerTracker(log logging.Logger) *peerTracker {
 
 // Returns true if we're not connected to enough peers.
 // Otherwise returns true probabilistically based on the number of tracked peers.
+// Assumes p.lock is held.
 func (p *peerTracker) shouldTrackNewPeer() bool {
 	numResponsivePeers := p.responsivePeers.Len()
 	if numResponsivePeers < desiredMinResponsivePeers {
@@ -105,6 +108,9 @@ func (p *peerTracker) shouldTrackNewPeer() bool {
 // Otherwise, with probability [randomPeerProbability] returns a random peer from [p.responsivePeers].
 // With probability [1-randomPeerProbability] returns the peer in [p.bandwidthHeap] with the highest bandwidth.
 func (p *peerTracker) GetAnyPeer(minVersion *version.Application) (ids.NodeID, bool) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
 	if p.shouldTrackNewPeer() {
 		for nodeID := range p.peers {
 			// if minVersion is specified and peer's version is less, skip
@@ -148,13 +154,19 @@ func (p *peerTracker) GetAnyPeer(minVersion *version.Application) (ids.NodeID, b
 
 // Record that we sent a request to [nodeID].
 func (p *peerTracker) TrackPeer(nodeID ids.NodeID) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
 	p.trackedPeers.Add(nodeID)
 	// p.numTrackedPeers.Set(float64(p.trackedPeers.Len()))
 }
 
 // Record that we observed that [nodeID]'s bandwidth is [bandwidth].
 // Adds the peer's bandwidth averager to the bandwidth heap.
 func (p *peerTracker) TrackBandwidth(nodeID ids.NodeID, bandwidth float64) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
 	peer := p.peers[nodeID]
 	if peer == nil {
 		// we're not connected to this peer, nothing to do here
@@ -184,6 +196,9 @@ func (p *peerTracker) TrackBandwidth(nodeID ids.NodeID, bandwidth float64) {
 
 // Connected should be called when [nodeID] connects to this node
 func (p *peerTracker) Connected(nodeID ids.NodeID, nodeVersion *version.Application) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
 	peer := p.peers[nodeID]
 	if peer == nil {
 		p.peers[nodeID] = &peerInfo{
@@ -216,6 +231,9 @@ func (p *peerTracker) Connected(nodeID ids.NodeID, nodeVersion *version.Applicat
 
 // Disconnected should be called when [nodeID] disconnects from this node
 func (p *peerTracker) Disconnected(nodeID ids.NodeID) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
 	p.bandwidthHeap.Remove(nodeID)
 	p.trackedPeers.Remove(nodeID)
 	// p.numTrackedPeers.Set(float64(p.trackedPeers.Len()))
@@ -226,5 +244,8 @@ func (p *peerTracker) Disconnected(nodeID ids.NodeID) {
 
 // Returns the number of peers the node is connected to.
 func (p *peerTracker) Size() int {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
 	return len(p.peers)
 }