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

The prefix handler ran one PeerDown per peer, one after another. When
the sessions of several peers went down together, their walks queued up.
A returning full-table peer's messages stayed held back while the walks
ahead of its own ran, and could reach the limit of 20,000.

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 since the previous one began. The doc of
maxHeldBackMessages says how long a peer's messages are held back. The
limit is now 150,000 messages, about 150 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,15 @@ 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, or the prefix itself.
+						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,28 @@ 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)}}},
 		},
+		{
+			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 +310,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)}}}
@@ -333,6 +340,32 @@ func TestPeerDownLeavesEveryPrefix(t *testing.T) {
 	}
 }
 
+// TestPeersDownTakesOutPeersOfDifferentPrefixes takes peerA and peerB out in
+// one PeersDown. peerA carries only prefixV4, and peerB only prefixV6. So
+// peerA's leaving removes prefixV4 before PeersDown looks for peerB there. Both
+// prefixes must then be gone.
+func TestPeersDownTakesOutPeersOfDifferentPrefixes(t *testing.T) {
+	t.Parallel()
+
+	table := routetable.New(routetable.Params{})
+	prefixes := []netip.Prefix{
+		netip.MustParsePrefix(prefixV4),
+		netip.MustParsePrefix(prefixV6),
+	}
+
+	table.Announce(prefixes[0], asA, routeOf(peerA, asA, 10))
+	table.Announce(prefixes[1], asB, routeOf(peerB, asB, 20))
+	table.PeersDown([]string{peerA, peerB})
+
+	for _, prefix := range prefixes {
+		got := table.Origins(prefix)
+		if got != nil {
+			t.Errorf("Origins(%s) = %+v after its only peer went down, want nil",
+				prefix, got)
+		}
+	}
+}
+
 func TestCountsOriginsWithoutKeptRoute(t *testing.T) {
 	t.Parallel()
 
@@ -358,7 +391,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 +497,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 +506,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 +650,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()) },