feat: simplify kademlia topology build up

pkg/topology/kademlia/kademlia.go

@@ -248,10 +248,20 @@ func (k *Kad) connectBalanced(wg *sync.WaitGroup, peerConnChan chan<- *peerConnI
 		return false
 	}
 
+	depth := k.NeighborhoodDepth()
+
 	for i := range k.commonBinPrefixes {
-		if i >= int(k.NeighborhoodDepth()) {
+
+		binPeersLength := k.knownPeers.BinSize(uint8(i))
+
+		// balancer should skip on bins where neighborhood connector would connect to peers anyway
+		// and there are not enough peers in known addresses to properly balance the bin
+		if i >= int(depth) && binPeersLength < len(k.commonBinPrefixes[i]) {
 			continue
 		}
+
+		binPeers := k.knownPeers.BinPeers(uint8(i))
+
 		for j := range k.commonBinPrefixes[i] {
 			pseudoAddr := k.commonBinPrefixes[i][j]
 
@@ -270,7 +280,7 @@ func (k *Kad) connectBalanced(wg *sync.WaitGroup, peerConnChan chan<- *peerConnI
 			}
 
 			// Connect to closest known peer which we haven't tried connecting to recently.
-			closestKnownPeer, err := closestPeer(k.knownPeers, pseudoAddr, skipPeers)
+			closestKnownPeer, err := closestPeerInSlice(binPeers, pseudoAddr, skipPeers)
 			if err != nil {
 				if errors.Is(err, topology.ErrNotFound) {
 					break
@@ -305,19 +315,22 @@ func (k *Kad) connectBalanced(wg *sync.WaitGroup, peerConnChan chan<- *peerConnI
 
 // connectNeighbours attempts to connect to the neighbours
 // which were not considered by the connectBalanced method.
-func (k *Kad) connectNeighbours(wg *sync.WaitGroup, peerConnChan, peerConnChan2 chan<- *peerConnInfo) {
-	const multiplePeerThreshold = 8
+func (k *Kad) connectNeighbours(wg *sync.WaitGroup, peerConnChan chan<- *peerConnInfo) {
 
 	sent := 0
+	var currentPo uint8 = 0
+
 	_ = k.knownPeers.EachBinRev(func(addr swarm.Address, po uint8) (bool, bool, error) {
 		depth := k.NeighborhoodDepth()
 
-		if depth > po || po >= depth+multiplePeerThreshold {
+		// out of depth, skip bin
+		if po < depth {
 			return false, true, nil
 		}
 
-		if len(k.connectedPeers.BinPeers(po)) >= overSaturationPeers-1 {
-			return false, true, nil
+		if po != currentPo {
+			currentPo = po
+			sent = 0
 		}
 
 		if k.connectedPeers.Exists(addr) {
@@ -343,49 +356,13 @@ func (k *Kad) connectNeighbours(wg *sync.WaitGroup, peerConnChan, peerConnChan2
 
 		// We want to sent number of attempts equal to saturationPeers
 		// in order to speed up the topology build.
-		next := sent == saturationPeers
-		if next {
-			sent = 0
-		}
-		return false, next, nil
-	})
-
-	_ = k.knownPeers.EachBinRev(func(addr swarm.Address, po uint8) (bool, bool, error) {
-		depth := k.NeighborhoodDepth()
-
-		if po < depth+multiplePeerThreshold {
-			return false, true, nil
-		}
-
-		if k.connectedPeers.Exists(addr) {
-			return false, false, nil
-		}
-
-		if k.waitNext.Waiting(addr) {
-			k.metrics.TotalBeforeExpireWaits.Inc()
-			return false, false, nil
-		}
-
-		select {
-		case <-k.quit:
-			return true, false, nil
-		default:
-			wg.Add(1)
-			peerConnChan2 <- &peerConnInfo{
-				po:   po,
-				addr: addr,
-			}
-		}
-
-		// The bin could be saturated or not, so a decision cannot
-		// be made before checking the next peer, so we iterate to next.
-		return false, true, nil
+		return false, sent == saturationPeers, nil
 	})
 }
 
 // connectionAttemptsHandler handles the connection attempts
 // to peers sent by the producers to the peerConnChan.
-func (k *Kad) connectionAttemptsHandler(ctx context.Context, wg *sync.WaitGroup, peerConnChan, peerConnChan2 <-chan *peerConnInfo) {
+func (k *Kad) connectionAttemptsHandler(ctx context.Context, wg *sync.WaitGroup, neighbourhoodChan, balanceChan <-chan *peerConnInfo) {
 	connect := func(peer *peerConnInfo) {
 		bzzAddr, err := k.addressBook.Get(peer.addr)
 		switch {
@@ -470,11 +447,11 @@ func (k *Kad) connectionAttemptsHandler(ctx context.Context, wg *sync.WaitGroup,
 			}
 		}
 	}
-	for i := 0; i < 32; i++ {
-		go connAttempt(peerConnChan)
+	for i := 0; i < 16; i++ {
+		go connAttempt(balanceChan)
 	}
-	for i := 0; i < 8; i++ {
-		go connAttempt(peerConnChan2)
+	for i := 0; i < 32; i++ {
+		go connAttempt(neighbourhoodChan)
 	}
 }
 
@@ -502,9 +479,9 @@ func (k *Kad) manage() {
 	// The wg makes sure that we wait for all the connection attempts,
 	// spun up by goroutines, to finish before we try the boot-nodes.
 	var wg sync.WaitGroup
-	peerConnChan := make(chan *peerConnInfo)
-	peerConnChan2 := make(chan *peerConnInfo)
-	go k.connectionAttemptsHandler(ctx, &wg, peerConnChan, peerConnChan2)
+	neighbourhoodChan := make(chan *peerConnInfo)
+	balanceChan := make(chan *peerConnInfo)
+	go k.connectionAttemptsHandler(ctx, &wg, neighbourhoodChan, balanceChan)
 
 	for {
 		select {
@@ -536,8 +513,8 @@ func (k *Kad) manage() {
 			}
 
 			oldDepth := k.NeighborhoodDepth()
-			k.connectNeighbours(&wg, peerConnChan, peerConnChan2)
-			k.connectBalanced(&wg, peerConnChan2)
+			k.connectBalanced(&wg, balanceChan)
+			k.connectNeighbours(&wg, neighbourhoodChan)
 			wg.Wait()
 
 			k.depthMu.Lock()
@@ -966,13 +943,46 @@ func (k *Kad) notifyPeerSig() {
 
 func closestPeer(peers *pslice.PSlice, addr swarm.Address, spf sanctionedPeerFunc) (swarm.Address, error) {
 	closest := swarm.ZeroAddress
-	err := peers.EachBinRev(func(peer swarm.Address, po uint8) (bool, bool, error) {
+	err := peers.EachBinRev(closestPeerFunc(&closest, addr, spf))
+	if err != nil {
+		return closest, err
+	}
+
+	// check if found
+	if closest.IsZero() {
+		return closest, topology.ErrNotFound
+	}
+
+	return closest, nil
+}
+
+func closestPeerInSlice(peers []swarm.Address, addr swarm.Address, spf sanctionedPeerFunc) (swarm.Address, error) {
+	closest := swarm.ZeroAddress
+	closestFunc := closestPeerFunc(&closest, addr, spf)
+
+	for _, peer := range peers {
+		_, _, err := closestFunc(peer, 0)
+		if err != nil {
+			return closest, err
+		}
+	}
+
+	// check if found
+	if closest.IsZero() {
+		return closest, topology.ErrNotFound
+	}
+
+	return closest, nil
+}
+
+func closestPeerFunc(closest *swarm.Address, addr swarm.Address, spf sanctionedPeerFunc) func(peer swarm.Address, po uint8) (bool, bool, error) {
+	return func(peer swarm.Address, po uint8) (bool, bool, error) {
 		// check whether peer is sanctioned
 		if spf(peer) {
 			return false, false, nil
 		}
 		if closest.IsZero() {
-			closest = peer
+			*closest = peer
 			return false, false, nil
 		}
 		dcmp, err := swarm.DistanceCmp(addr.Bytes(), closest.Bytes(), peer.Bytes())
@@ -984,23 +994,13 @@ func closestPeer(peers *pslice.PSlice, addr swarm.Address, spf sanctionedPeerFun
 			// do nothing
 		case -1:
 			// current peer is closer
-			closest = peer
+			*closest = peer
 		case 1:
 			// closest is already closer to chunk
 			// do nothing
 		}
 		return false, false, nil
-	})
-	if err != nil {
-		return swarm.ZeroAddress, err
 	}
-
-	// check if found
-	if closest.IsZero() {
-		return swarm.ZeroAddress, topology.ErrNotFound
-	}
-
-	return closest, nil
 }
 
 func isIn(a swarm.Address, addresses []p2p.Peer) bool {

pkg/topology/pslice/pslice.go

@@ -96,6 +96,26 @@ func (s *PSlice) EachBinRev(pf topology.EachPeerFunc) error {
 	return nil
 }
 
+func (s *PSlice) BinSize(bin uint8) int {
+
+	s.RLock()
+	defer s.RUnlock()
+
+	b := int(bin)
+	if b >= len(s.bins) {
+		return 0
+	}
+
+	var bEnd int
+	if b == len(s.bins)-1 {
+		bEnd = len(s.peers)
+	} else {
+		bEnd = int(s.bins[b+1])
+	}
+
+	return bEnd - int(s.bins[b])
+}
+
 func (s *PSlice) BinPeers(bin uint8) []swarm.Address {
 	s.RLock()
 	defer s.RUnlock()

pkg/topology/pslice/pslice_test.go

@@ -284,6 +284,9 @@ func TestBinPeers(t *testing.T) {
 				if !isEqual(binPeers[bin], ps.BinPeers(uint8(bin))) {
 					t.Fatal("peers list do not match")
 				}
+				if len(binPeers[bin]) != ps.BinSize(uint8(bin)) {
+					t.Fatal("peers list lengths do not match")
+				}
 			}
 
 			// out of bound bin check