Take every down peer out of the table in one walk (closes #225)

The prefix handler ran one PeerDown per peer, one after another, and
stopped taking messages at 20,000 held-back messages. Sessions going
down together queued up their walks, so a returning full-table peer
could reach that limit.

The table's PeersDown now takes several peers out of every prefix in
one walk. One walk runs at a time. The next one takes every peer whose
session went down meanwhile, so a message is held back for about two
walks at most. The limit is now 100,000 messages, about 100 MiB. The
status page shows how many messages each handler holds back.

Model: opus-5-5

internal/routetable/reads_test.go

@@ -227,7 +227,7 @@ func TestCounts(t *testing.T) {
 	})
 
 	// peerC was the last peer of the IPv6 prefix. It keeps its number.
-	table.PeerDown(peerC)
+	table.PeersDown([]string{peerC})
 	checkCounts(t, table, routetable.Counts{
 		Pairs:                2,
 		Peers:                3,

internal/routetable/routetable.go

@@ -28,7 +28,7 @@ const (
 	keptTimesDegree = 32
 
 	// walkBatch is how many prefixes a walk over the whole table visits in
-	// one hold of the lock. Overview and PeerDown each walk the whole table.
+	// one hold of the lock. Overview and PeersDown each walk the whole table.
 	// Callers waiting for the lock get it between batches.
 	walkBatch = 10000
 
@@ -205,20 +205,29 @@ func (t *Table) Withdraw(prefix netip.Prefix, peer string) {
 	}
 }
 
-// PeerDown applies a peer's session going down: the peer leaves every peer
-// set, and every route of the peer is dropped. When the peer carries any
-// prefix, PeerDown visits every prefix in the table, in batches of walkBatch.
-// After each batch, it lets the table's lock go for as long as it held the
-// lock for the batch. During a walk over many batches, reads and updates can
-// then take the lock about half of the time. An announcement by the peer
-// applied during the walk may be dropped too. So the caller applies the peer's
-// later updates only once PeerDown has returned.
-func (t *Table) PeerDown(peer string) {
+// PeersDown applies the sessions of peers going down: each of the peers leaves
+// every peer set, and every route of each is dropped. When any of them carries
+// a prefix, PeersDown visits every prefix in the table once, in batches of
+// walkBatch. After each batch, it lets the table's lock go for as long as it
+// held the lock for the batch. During a walk over many batches, reads and
+// updates can then take the lock about half of the time. An announcement by
+// one of the peers applied during the walk may be dropped too. So the caller
+// applies those peers' later updates only once PeersDown has returned.
+func (t *Table) PeersDown(peers []string) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
-	number, known := t.peerNumbers[peer]
-	if !known || t.peerPrefixes[number] == 0 {
+	// numbers holds the number of each of the peers that carries a prefix.
+	numbers := make([]uint, 0, len(peers))
+
+	for _, peer := range peers {
+		number, known := t.peerNumbers[peer]
+		if known && t.peerPrefixes[number] > 0 {
+			numbers = append(numbers, number)
+		}
+	}
+
+	if len(numbers) == 0 {
 		return
 	}
 
@@ -228,9 +237,14 @@ func (t *Table) PeerDown(peer string) {
 	for _, f := range []*family{&t.ipv4, &t.ipv6} {
 		for _, prefixes := range f.byLength {
 			for prefix, origins := range prefixes {
-				o := originWithPeer(origins, number)
-				if o != nil {
-					t.leave(prefix, o, number)
+				for _, number := range numbers {
+					o := originWithPeer(origins, number)
+					if o != nil {
+						t.leave(prefix, o, number)
+
+						// leave may have removed an origin of the prefix.
+						origins = prefixes[prefix]
+					}
 				}
 
 				// Go allows the map to change before the next step of

internal/routetable/routetable_internal_test.go

@@ -7,11 +7,11 @@ import (
 	"time"
 )
 
-// TestPeerDownLetsReadsIn reads the table's counts again and again while a peer
-// that carries more than two batches of prefixes goes down. PeerDown lets the
-// lock go after every walkBatch prefixes, so a read must get in while the peer
-// still carries some of its prefixes but no longer all.
-func TestPeerDownLetsReadsIn(t *testing.T) {
+// TestPeersDownLetsReadsIn reads the table's counts again and again while a
+// peer that carries more than two batches of prefixes goes down. PeersDown lets
+// the lock go after every walkBatch prefixes, so a read must get in while the
+// peer still carries some of its prefixes but no longer all.
+func TestPeersDownLetsReadsIn(t *testing.T) {
 	t.Parallel()
 
 	const (
@@ -32,8 +32,8 @@ func TestPeerDownLetsReadsIn(t *testing.T) {
 		addr = addr.Next()
 	}
 
-	// Hold a read lock until PeerDown waits for the lock, so that the first
-	// read below starts once PeerDown holds it. TryRLock fails once a writer
+	// Hold a read lock until PeersDown waits for the lock, so that the first
+	// read below starts once PeersDown holds it. TryRLock fails once a writer
 	// waits.
 	table.mu.RLock()
 
@@ -42,7 +42,7 @@ func TestPeerDownLetsReadsIn(t *testing.T) {
 	go func() {
 		defer close(peerDownDone)
 
-		table.PeerDown(peer)
+		table.PeersDown([]string{peer})
 	}()
 
 	for table.mu.TryRLock() {
@@ -60,7 +60,7 @@ func TestPeerDownLetsReadsIn(t *testing.T) {
 
 		select {
 		case <-peerDownDone:
-			t.Fatal("no read got in while PeerDown ran")
+			t.Fatal("no read got in while PeersDown ran")
 		default:
 		}
 	}

internal/routetable/routetable_test.go

@@ -69,10 +69,10 @@ func withdraw(peer string) step {
 	}
 }
 
-// peerDown is the step of peer's session going down.
-func peerDown(peer string) step {
+// peersDown is the step of the sessions of peers going down.
+func peersDown(peers ...string) step {
 	return func(table *routetable.Table, _ netip.Prefix) {
-		table.PeerDown(peer)
+		table.PeersDown(peers)
 	}
 }
 
@@ -272,21 +272,29 @@ func TestWithdrawalsAndPeerDowns(t *testing.T) {
 		{
 			name: "a peer going down leaves the prefix",
 			steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
-				peerDown(peerB)},
+				peersDown(peerB)},
 			want: []routetable.Origin{{ASN: asA, Peers: []string{peerA},
 				Routes: []routetable.Route{routeOf(peerA, asA, 10)}}},
 		},
 		{
 			name: "a peer going down after it withdrew changes nothing",
 			steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
-				withdraw(peerB), peerDown(peerB)},
+				withdraw(peerB), peersDown(peerB)},
 			want: []routetable.Origin{{ASN: asA, Peers: []string{peerA},
 				Routes: []routetable.Route{routeOf(peerA, asA, 10)}}},
 		},
+		{
+			// peerA's leaving removes the origin asA, before peerB leaves.
+			name: "peers going down together each leave the prefix",
+			steps: []step{announce(peerA, asA, 10), announce(peerB, asB, 20),
+				announce(peerC, asB, 30), peersDown(peerA, peerB)},
+			want: []routetable.Origin{{ASN: asB, Peers: []string{peerC},
+				Routes: []routetable.Route{routeOf(peerC, asB, 30)}}},
+		},
 	})
 }
 
-func TestPeerDownLeavesEveryPrefix(t *testing.T) {
+func TestPeersDownLeavesEveryPrefix(t *testing.T) {
 	t.Parallel()
 
 	table := routetable.New(routetable.Params{})
@@ -303,7 +311,7 @@ func TestPeerDownLeavesEveryPrefix(t *testing.T) {
 	}
 
 	table.Announce(prefixes[0], asB, routeOf(peerB, asB, 20))
-	table.PeerDown(peerA)
+	table.PeersDown([]string{peerA})
 
 	want := []routetable.Origin{{ASN: asB, Peers: []string{peerB},
 		Routes: []routetable.Route{routeOf(peerB, asB, 20)}}}
@@ -358,7 +366,7 @@ func TestCountsOriginsWithoutKeptRoute(t *testing.T) {
 		{announce(peerC, asA, 30), 0},
 		{withdraw(peerB), 0},
 		{withdraw(peerC), 1},
-		{peerDown(peerA), 0},
+		{peersDown(peerA), 0},
 	}
 
 	for i, s := range steps {
@@ -464,7 +472,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
 	checkChanges(t, table, []change{}, map[string]uint{})
 
 	// peerA going down changes v4, the one prefix it carries.
-	table.PeerDown(peerA)
+	table.PeersDown([]string{peerA})
 	checkChanges(t, table, []change{
 		{Prefix: prefixV4, OriginASN: asA, Peers: []uint{1, 2},
 			Routes: []routetable.Route{
@@ -473,7 +481,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
 	}, map[string]uint{})
 
 	// peerA, which carries nothing now, goes down again.
-	table.PeerDown(peerA)
+	table.PeersDown([]string{peerA})
 	checkChanges(t, table, []change{}, map[string]uint{})
 }
 
@@ -617,7 +625,7 @@ func TestConcurrentUse(t *testing.T) {
 		func(round int) { table.Announce(prefix, asA, routeOf(peerA, asA, round)) },
 		func(round int) { table.Announce(prefix, asB, routeOf(peerB, asB, round)) },
 		func(int) { table.Withdraw(prefix, peerA) },
-		func(int) { table.PeerDown(peerB) },
+		func(int) { table.PeersDown([]string{peerB}) },
 		func(int) { table.PutBackChanges(table.TakeChanges(math.MaxInt)) },
 		func(int) { table.ChangeCount() },
 		func(int) { table.Lookup(prefix.Addr()) },