avoid performing garbage collect on each topology update

gossip_test.go

@@ -156,6 +156,7 @@ func TestGossipTopology(t *testing.T) {
 	// Drop the connection from 1 to 3
 	r1.DeleteTestGossipConnection(r3)
 	sendPendingGossip(r1, r2, r3)
+	forcePendingGC(r1, r2, r3)
 	checkTopology(t, r1, r1.tp(r2), r2.tp(r1))
 	checkTopology(t, r2, r1.tp(r2), r2.tp(r1))
 	// r3 still thinks r1 has a connection to it

peer_test.go

@@ -13,3 +13,16 @@ func TestPeerRoutes(t *testing.T) {
 func TestPeerForEachConnectedPeer(t *testing.T) {
 	t.Skip("TODO")
 }
+
+func forcePendingGC(routers ...*Router) {
+	for _, router := range routers {
+		router.Peers.Lock()
+		if router.Peers.pendingGC {
+			var pending peersPendingNotifications
+			router.Peers.garbageCollect(&pending)
+			router.Peers.unlockAndNotify(&pending)
+		} else {
+			router.Peers.Unlock()
+		}
+	}
+}

peers.go

@@ -6,6 +6,11 @@ import (
 	"io"
 	"math/rand"
 	"sync"
+	"time"
+)
+
+const (
+	gcInterval = 1 * time.Second
 )
 
 // Peers collects all of the known peers in the mesh, including ourself.
@@ -18,6 +23,8 @@ type Peers struct {
 
 	// Called when the mapping from short IDs to peers changes
 	onInvalidateShortIDs []func()
+	timer                *time.Timer
+	pendingGC            bool
 }
 
 type shortIDPeers struct {
@@ -60,8 +67,11 @@ func newPeers(ourself *localPeer) *Peers {
 		ourself:   ourself,
 		byName:    make(map[PeerName]*Peer),
 		byShortID: make(map[PeerShortID]shortIDPeers),
+		timer:     time.NewTimer(gcInterval),
 	}
 	peers.fetchWithDefault(ourself.Peer)
+	peers.timer.Stop()
+	go peers.actorLoop()
 	return peers
 }
 
@@ -339,6 +349,18 @@ func (peers *Peers) forEach(fun func(*Peer)) {
 	}
 }
 
+func (peers *Peers) actorLoop() {
+	for {
+		select {
+		case <-peers.timer.C:
+			peers.GarbageCollect()
+			peers.Lock()
+			peers.pendingGC = false
+			peers.Unlock()
+		}
+	}
+}
+
 // Merge an incoming update with our own topology.
 //
 // We add peers hitherto unknown to us, and update peers for which the
@@ -363,7 +385,6 @@ func (peers *Peers) applyUpdate(update []byte) (peerNameSet, peerNameSet, error)
 
 	// Now apply the updates
 	newUpdate := peers.applyDecodedUpdate(decodedUpdate, decodedConns, &pending)
-	peers.garbageCollect(&pending)
 	for _, peerRemoved := range pending.removed {
 		delete(newUpdate, peerRemoved.Name)
 	}
@@ -373,6 +394,13 @@ func (peers *Peers) applyUpdate(update []byte) (peerNameSet, peerNameSet, error)
 		updateNames[peer.Name] = struct{}{}
 	}
 
+	if !peers.pendingGC {
+		// schedule a GarbageCollect() to run after gcInterval time period
+		// corresponding to all topology updates received during the period
+		peers.timer.Reset(gcInterval)
+		peers.pendingGC = true
+	}
+
 	return updateNames, newUpdate, nil
 }