Write the routing table every second and read it at startup (part of #122)
check / check (push) Canceled after 0s
check / check (push) Canceled after 0s
The prefix handler applies each update to the routing table in memory as it handles it, sorted into announcements and withdrawals as for the live route tables. Each flush writes the table's changes first. A failed write puts its changes and new peers back in the table, so the next write takes them with each pair's content then. Stop writes the last changes, and logs those left when its deadline passes. Startup reads the table before the streamer connects, and logs its pairs, peers and time. The per-peer counts are a slice, so the read takes about a third less time. The health check's start period is 60 seconds, so a full table's read fits before upaas looks. Model: opus-5-5
This commit is contained in:
@@ -31,6 +31,11 @@ type OriginatedPrefix struct {
|
||||
|
||||
// Counts are the table's figures for the stats.
|
||||
type Counts struct {
|
||||
// Pairs counts the prefix and origin AS pairs that some peer carries.
|
||||
Pairs int
|
||||
// Peers counts the peers that have a number: those given in
|
||||
// Params.PeerNumbers and those numbered since.
|
||||
Peers int
|
||||
IPv4Prefixes int
|
||||
IPv6Prefixes int
|
||||
// IPv4PrefixesByLength[n] and IPv6PrefixesByLength[n] are how many
|
||||
@@ -160,6 +165,7 @@ func (t *Table) Counts() Counts {
|
||||
defer t.mu.RUnlock()
|
||||
|
||||
counts := Counts{
|
||||
Peers: len(t.peerNumbers),
|
||||
IPv4PrefixesByLength: t.ipv4.prefixesByLength(),
|
||||
IPv6PrefixesByLength: t.ipv6.prefixesByLength(),
|
||||
IPv4Routes: t.ipv4.routes,
|
||||
@@ -175,6 +181,10 @@ func (t *Table) Counts() Counts {
|
||||
counts.IPv6Prefixes += n
|
||||
}
|
||||
|
||||
for _, prefixes := range t.byOrigin {
|
||||
counts.Pairs += len(prefixes)
|
||||
}
|
||||
|
||||
oldest, found := t.keptTimes.Min()
|
||||
if found {
|
||||
counts.OldestKeptRoute = &oldest.time
|
||||
|
||||
@@ -194,6 +194,8 @@ func TestCounts(t *testing.T) {
|
||||
announceAll(table, "2001:db8::/32", asA, 5, peerA)
|
||||
announceAll(table, "2001:db8::/32", asA, 25, peerC)
|
||||
checkCounts(t, table, routetable.Counts{
|
||||
Pairs: 3,
|
||||
Peers: 3,
|
||||
IPv4Prefixes: 2,
|
||||
IPv6Prefixes: 1,
|
||||
IPv4PrefixesByLength: byLength(ipv4Bits, map[int]int{16: 1, 24: 1}),
|
||||
@@ -209,6 +211,8 @@ func TestCounts(t *testing.T) {
|
||||
announceAll(table, prefixV4, asA, 30, peerC)
|
||||
table.Withdraw(netip.MustParsePrefix("2001:db8::/32"), peerA)
|
||||
checkCounts(t, table, routetable.Counts{
|
||||
Pairs: 3,
|
||||
Peers: 3,
|
||||
IPv4Prefixes: 2,
|
||||
IPv6Prefixes: 1,
|
||||
IPv4PrefixesByLength: byLength(ipv4Bits, map[int]int{16: 1, 24: 1}),
|
||||
@@ -219,9 +223,11 @@ func TestCounts(t *testing.T) {
|
||||
NewestKeptRoute: timeAt(30),
|
||||
})
|
||||
|
||||
// peerC was the last peer of the IPv6 prefix.
|
||||
// peerC was the last peer of the IPv6 prefix. It keeps its number.
|
||||
table.PeerDown(peerC)
|
||||
checkCounts(t, table, routetable.Counts{
|
||||
Pairs: 2,
|
||||
Peers: 3,
|
||||
IPv4Prefixes: 2,
|
||||
IPv4PrefixesByLength: byLength(ipv4Bits, map[int]int{16: 1, 24: 1}),
|
||||
IPv6PrefixesByLength: byLength(ipv6Bits, nil),
|
||||
|
||||
@@ -7,6 +7,7 @@ package routetable
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"maps"
|
||||
"net/netip"
|
||||
"slices"
|
||||
"sync"
|
||||
@@ -102,8 +103,11 @@ type Table struct {
|
||||
peerNumbers map[string]uint
|
||||
peerAddresses map[uint]string
|
||||
nextPeerNumber uint
|
||||
// peerPrefixes counts the prefixes each peer carries, by its number.
|
||||
peerPrefixes map[uint]int
|
||||
// peerPrefixes[n] counts the prefixes the peer numbered n carries, for
|
||||
// each n below nextPeerNumber. It is a slice, not a map, because Load
|
||||
// counts once for every peer of every pair it loads: with a map, reading
|
||||
// a full table at startup took several times as long.
|
||||
peerPrefixes []int
|
||||
|
||||
// newPeers and changed are what TakeChanges hands over next: the peers
|
||||
// numbered, and the prefix and origin AS pairs changed, since it last ran.
|
||||
@@ -127,7 +131,6 @@ func New(params Params) *Table {
|
||||
byOrigin: make(map[int]map[netip.Prefix]*origin),
|
||||
peerNumbers: make(map[string]uint, len(params.PeerNumbers)),
|
||||
peerAddresses: make(map[uint]string, len(params.PeerNumbers)),
|
||||
peerPrefixes: make(map[uint]int),
|
||||
newPeers: make(map[string]uint),
|
||||
changed: make(map[pair]struct{}),
|
||||
keptTimes: btree.NewG(keptTimesDegree, func(a, b keptTime) bool {
|
||||
@@ -141,6 +144,8 @@ func New(params Params) *Table {
|
||||
t.nextPeerNumber = max(t.nextPeerNumber, number+1)
|
||||
}
|
||||
|
||||
t.peerPrefixes = make([]int, t.nextPeerNumber)
|
||||
|
||||
return t
|
||||
}
|
||||
|
||||
@@ -240,13 +245,14 @@ func (t *Table) Load(change Change) {
|
||||
o := t.findOrAddOrigin(prefix, change.OriginASN)
|
||||
wasWithoutKeptRoute := o.withoutKeptRoute()
|
||||
|
||||
for peer := range change.Peers.EachSet() {
|
||||
if !o.peers.Test(peer) {
|
||||
o.peers.Set(peer)
|
||||
// The peers not yet in the peer set join it in one union, which takes
|
||||
// less time than setting their bits one at a time.
|
||||
joining := change.Peers.Difference(&o.peers)
|
||||
o.peers.InPlaceUnion(joining)
|
||||
t.familyOf(prefix.Addr()).routes += int(joining.Count())
|
||||
|
||||
t.familyOf(prefix.Addr()).routes++
|
||||
t.peerPrefixes[peer]++
|
||||
}
|
||||
for peer := range joining.EachSet() {
|
||||
t.peerPrefixes[peer]++
|
||||
}
|
||||
|
||||
// Oldest first, as they were announced, so that two routes with the same
|
||||
@@ -289,6 +295,30 @@ func (t *Table) TakeChanges() ([]Change, map[string]uint) {
|
||||
return changes, newPeers
|
||||
}
|
||||
|
||||
// PutBackChanges records again what one TakeChanges call returned, for a
|
||||
// caller that could not write it. The next TakeChanges then returns those pairs
|
||||
// and peers again, with the pairs that changed since. It returns each pair with
|
||||
// its content at that time, so a later change of a pair replaces the earlier.
|
||||
func (t *Table) PutBackChanges(changes []Change, newPeers map[string]uint) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
|
||||
for _, change := range changes {
|
||||
t.changed[pair{prefix: change.Prefix, asn: change.OriginASN}] = struct{}{}
|
||||
}
|
||||
|
||||
maps.Copy(t.newPeers, newPeers)
|
||||
}
|
||||
|
||||
// ChangeCount returns how many prefix and origin AS pairs changed since
|
||||
// TakeChanges last ran: how many it would return now.
|
||||
func (t *Table) ChangeCount() int {
|
||||
t.mu.RLock()
|
||||
defer t.mu.RUnlock()
|
||||
|
||||
return len(t.changed)
|
||||
}
|
||||
|
||||
// numberPeer returns the peer's number, and gives it the next one the first
|
||||
// time the table sees it.
|
||||
func (t *Table) numberPeer(address string) uint {
|
||||
@@ -301,6 +331,7 @@ func (t *Table) numberPeer(address string) uint {
|
||||
t.nextPeerNumber++
|
||||
t.peerNumbers[address] = number
|
||||
t.peerAddresses[number] = address
|
||||
t.peerPrefixes = append(t.peerPrefixes, 0)
|
||||
t.newPeers[address] = number
|
||||
|
||||
return number
|
||||
|
||||
@@ -476,6 +476,43 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
|
||||
checkChanges(t, table, []change{}, map[string]uint{})
|
||||
}
|
||||
|
||||
// TestPutBackChanges takes a table's changes and puts them back, as a caller
|
||||
// does whose write failed. The next TakeChanges must return them again, with
|
||||
// each pair's content then, and with the changes since.
|
||||
func TestPutBackChanges(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
table := routetable.New(routetable.Params{})
|
||||
v4 := netip.MustParsePrefix(prefixV4)
|
||||
v6 := netip.MustParsePrefix(prefixV6)
|
||||
|
||||
table.Announce(v4, asA, routeOf(peerA, asA, 10))
|
||||
table.Announce(v6, asA, routeOf(peerA, asA, 10))
|
||||
table.PutBackChanges(table.TakeChanges())
|
||||
|
||||
if got := table.ChangeCount(); got != 2 {
|
||||
t.Errorf("ChangeCount() = %d after the changes were put back, want 2",
|
||||
got)
|
||||
}
|
||||
|
||||
// peerA withdraws v4, so v4's earlier content must not come back; peerB
|
||||
// announces v6, and is a new peer as well.
|
||||
table.Withdraw(v4, peerA)
|
||||
table.Announce(v6, asA, routeOf(peerB, asA, 20))
|
||||
|
||||
checkChanges(t, table, []change{
|
||||
{Prefix: prefixV4, OriginASN: asA, Removed: true},
|
||||
{Prefix: prefixV6, OriginASN: asA, Peers: []uint{0, 1},
|
||||
Routes: []routetable.Route{
|
||||
routeOf(peerB, asA, 20), routeOf(peerA, asA, 10),
|
||||
}},
|
||||
}, map[string]uint{peerA: 0, peerB: 1})
|
||||
|
||||
if got := table.ChangeCount(); got != 0 {
|
||||
t.Errorf("ChangeCount() = %d after TakeChanges, want 0", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoad(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -543,7 +580,8 @@ func TestConcurrentUse(t *testing.T) {
|
||||
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.TakeChanges() },
|
||||
func(int) { table.PutBackChanges(table.TakeChanges()) },
|
||||
func(int) { table.ChangeCount() },
|
||||
func(int) { table.Lookup(prefix.Addr()) },
|
||||
func(int) { table.Origins(prefix) },
|
||||
func(int) { table.OriginatedPrefixes(asA) },
|
||||
|
||||
Reference in New Issue
Block a user