fix refreshing buckets during query. getting topK peers is more efficient.

kpeerset/sorted_peerset.go

@@ -116,23 +116,24 @@ func (ps *SortedPeerset) Add(p peer.ID) {
 }
 
 func (ps *SortedPeerset) TopK() []peer.ID {
-	return ps.getTopK(nil)
+	ps.lock.Lock()
+	defer ps.lock.Unlock()
+
+	topK := make([]peer.ID, 0, len(ps.heapTopKPeers.data))
+	for _, pm := range ps.heapTopKPeers.data {
+		topK = append(topK, pm.Peer())
+	}
+
+	return topK
 }
 
 func (ps *SortedPeerset) KUnqueried() []peer.ID {
-	return ps.getTopK(func(p peer.ID) bool {
-		_, ok := ps.queried[p]
-		return ok
-	})
-}
-
-func (ps *SortedPeerset) getTopK(filter func(p peer.ID) bool) []peer.ID {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
 
 	topK := make([]IPeerMetric, 0, len(ps.heapTopKPeers.data))
 	for _, pm := range ps.heapTopKPeers.data {
-		if filter == nil || !filter(pm.Peer()) {
+		if _, ok := ps.queried[pm.Peer()]; !ok {
 			topK = append(topK, pm.IPeerMetric)
 		}
 	}

lookup.go

@@ -117,7 +117,7 @@ func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key string) (<-chan pee
 		out <- p
 	}
 
-	if ctx.Err() != nil {
+	if ctx.Err() == nil {
 		// refresh the cpl for this key as the query was successful
 		dht.routingTable.ResetCplRefreshedAtForID(kadID, time.Now())
 	}

routing.go

@@ -403,7 +403,7 @@ func (dht *IpfsDHT) getValues(ctx context.Context, key string, stopFn func() boo
 			}
 		}
 
-		if !shortcutTaken {
+		if !shortcutTaken && ctx.Err() == nil {
 			kadID := kb.ConvertKey(key)
 			// refresh the cpl for this key as the query was successful
 			dht.routingTable.ResetCplRefreshedAtForID(kadID, time.Now())
@@ -678,7 +678,7 @@ func (dht *IpfsDHT) FindPeer(ctx context.Context, id peer.ID) (_ peer.AddrInfo,
 
 		return dht.peerstore.PeerInfo(id), nil
 	} else {
-		if ctx.Err() != nil {
+		if ctx.Err() == nil {
 			kadID := kb.ConvertPeerID(id)
 			// refresh the cpl for this key as the query was successful
 			dht.routingTable.ResetCplRefreshedAtForID(kadID, time.Now())