Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9e7e87b68d |
No files matched your search
@@ -445,9 +445,8 @@ the routing table, and everything else from the database.
|
|||||||
- A withdrawal removes the peer's route. So does an announcement whose
|
- A withdrawal removes the peer's route. So does an announcement whose
|
||||||
AS path is empty or holds AS 0 anywhere.
|
AS path is empty or holds AS 0 anywhere.
|
||||||
- A peer whose BGP session goes down leaves every prefix. The handler
|
- A peer whose BGP session goes down leaves every prefix. The handler
|
||||||
holds back the peer's later updates. It applies them in order after
|
holds back the peer's later updates until the peer has left every
|
||||||
the peer has left every prefix and after the held-back updates of
|
prefix, then applies them in order.
|
||||||
every peer whose session went down before it.
|
|
||||||
|
|
||||||
Every second it writes the changed prefix and origin AS pairs to the
|
Every second it writes the changed prefix and origin AS pairs to the
|
||||||
`routes` table. A write takes at most 20,000 pairs, and the rest follow
|
`routes` table. A write takes at most 20,000 pairs, and the rest follow
|
||||||
@@ -652,11 +651,10 @@ What happens at each limit:
|
|||||||
at most about 0.1 GiB of the Go heap.
|
at most about 0.1 GiB of the Go heap.
|
||||||
|
|
||||||
- Held-back messages: the prefix handler holds back a peer's messages after the
|
- Held-back messages: the prefix handler holds back a peer's messages after the
|
||||||
peer's session goes down. It applies them after the peer has left every prefix
|
peer's session goes down. It applies them once the peer has left every prefix.
|
||||||
and after the held-back messages of every peer whose session went down before
|
It holds at most 20,000 such messages (`maxHeldBackMessages`), about 20 MiB.
|
||||||
it. It holds at most 150,000 such messages (`maxHeldBackMessages`), about 150
|
At that number it takes nothing more off its queue until it has applied one of
|
||||||
MiB. At that number it takes nothing more off its queue until it has applied
|
them.
|
||||||
one of them.
|
|
||||||
|
|
||||||
With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds
|
With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds
|
||||||
with the goroutine count and Go memory figures.
|
with the goroutine count and Go memory figures.
|
||||||
|
|||||||
@@ -22,7 +22,7 @@ formatting of the Go code and the markdown. Then it builds the image.
|
|||||||
The image sets memory ceilings for a 5 GiB container (README "Memory"). README
|
The image sets memory ceilings for a 5 GiB container (README "Memory"). README
|
||||||
"Running under upaas" says how to run it.
|
"Running under upaas" says how to run it.
|
||||||
|
|
||||||
A five-day run of `1c65656` on the live feed, from 2026-10-27 to 2026-11-01,
|
A five-day run of `1c65656` on the live feed, from 27 October to 1 November,
|
||||||
found three problems:
|
found three problems:
|
||||||
|
|
||||||
- The prefix handler dropped updates while a peer's session went down
|
- The prefix handler dropped updates while a peer's session went down
|
||||||
@@ -34,15 +34,12 @@ found three problems:
|
|||||||
|
|
||||||
All three are fixed on `next`. So is
|
All three are fixed on `next`. So is
|
||||||
https://git.eeqj.de/sneak/routewatch/issues/222, which the third fix brought up.
|
https://git.eeqj.de/sneak/routewatch/issues/222, which the third fix brought up.
|
||||||
A run of `26d939a` on the live feed showed the prefix handler still dropping
|
A 24-hour run of `26d939a` on the live feed is checking whether any handler
|
||||||
updates while the home page's cards were worked out.
|
still drops updates.
|
||||||
https://git.eeqj.de/sneak/routewatch/issues/226 fixed that on `next`.
|
|
||||||
https://git.eeqj.de/sneak/routewatch/issues/225, also on `next`, takes the peers
|
|
||||||
whose sessions go down together out of the table in one walk. A 24-hour run of
|
|
||||||
`9bd0e91` on the live feed is checking whether any handler still drops updates.
|
|
||||||
|
|
||||||
The README rewrite, https://git.eeqj.de/sneak/routewatch/pulls/210, is being
|
The README rewrite, https://git.eeqj.de/sneak/routewatch/pulls/210, has failed
|
||||||
reworked. It is part of https://git.eeqj.de/sneak/routewatch/issues/157.
|
three reviews. It is paused until sneak has reviewed it. It is part of
|
||||||
|
https://git.eeqj.de/sneak/routewatch/issues/157.
|
||||||
|
|
||||||
Two questions wait for sneak: which history the prefix page keeps
|
Two questions wait for sneak: which history the prefix page keeps
|
||||||
(https://git.eeqj.de/sneak/routewatch/issues/161), and which fields the IP
|
(https://git.eeqj.de/sneak/routewatch/issues/161), and which fields the IP
|
||||||
@@ -53,8 +50,8 @@ https://git.eeqj.de/sneak/routewatch/issues/86.
|
|||||||
|
|
||||||
# Next Step
|
# Next Step
|
||||||
|
|
||||||
Post the outcome of the 24-hour run of `9bd0e91` on
|
Post the outcome of the 24-hour run of `26d939a` on
|
||||||
https://git.eeqj.de/sneak/routewatch/issues/226: whether any handler still drops
|
https://git.eeqj.de/sneak/routewatch/issues/216: whether any handler still drops
|
||||||
updates, and the process's peak memory against 5 GiB. File an issue for each
|
updates, and the process's peak memory against 5 GiB. File an issue for each
|
||||||
problem the run finds.
|
problem the run finds.
|
||||||
|
|
||||||
@@ -64,20 +61,6 @@ problem the run finds.
|
|||||||
the rules of https://git.eeqj.de/sneak/routewatch/issues/157. Every Completed
|
the rules of https://git.eeqj.de/sneak/routewatch/issues/157. Every Completed
|
||||||
Steps entry keeps its date, its facts and its issue. Status and Next Step
|
Steps entry keeps its date, its facts and its issue. Status and Next Step
|
||||||
describe the repo as the tracker shows it (part of #157)
|
describe the repo as the tracker shows it (part of #157)
|
||||||
- 2026-11-02: when the sessions of several peers go down, the prefix handler
|
|
||||||
takes the peers out of the routing table together. The table's `PeersDown`
|
|
||||||
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. Each peer used to have a walk of its own. A peer's messages stayed held
|
|
||||||
back while the walks queued ahead of its own ran. The handler holds back at
|
|
||||||
most 150,000 messages, about 150 MiB. It used to hold back at most 20,000. The
|
|
||||||
status page shows how many messages the handler holds back (closes #225)
|
|
||||||
- 2026-11-02: working out the home page's cards no longer stops the prefix
|
|
||||||
handler. The walk over the routing table holds the table's lock for 10,000
|
|
||||||
prefixes at a time, then lets the lock go for as long again, as `PeerDown`
|
|
||||||
does. It used to take the lock back at once. The prefix handler then applied
|
|
||||||
about one update per 10,000 prefixes walked. During a walk, its queue could
|
|
||||||
fill, and the streamer then dropped updates for it (closes #226)
|
|
||||||
- 2026-11-02: A skipped WHOIS lookup now makes the fetcher wait longer before
|
- 2026-11-02: A skipped WHOIS lookup now makes the fetcher wait longer before
|
||||||
the next one, as a failed lookup does. A lookup is skipped when its RIR is
|
the next one, as a failed lookup does. A lookup is skipped when its RIR is
|
||||||
paused. An RIR is one of the five regional Internet registries. The longest
|
paused. An RIR is one of the five regional Internet registries. The longest
|
||||||
@@ -425,6 +408,7 @@ problem the run finds.
|
|||||||
|
|
||||||
# Future Steps
|
# Future Steps
|
||||||
|
|
||||||
- Merge https://git.eeqj.de/sneak/routewatch/pulls/210 once a review passes. As
|
- Finish https://git.eeqj.de/sneak/routewatch/pulls/210 as sneak's review
|
||||||
the last of the two PRs for https://git.eeqj.de/sneak/routewatch/issues/157 to
|
directs. It merges once a review passes. Of the two PRs for
|
||||||
merge, it closes the issue.
|
https://git.eeqj.de/sneak/routewatch/issues/157, the one that merges last
|
||||||
|
closes the issue.
|
||||||
@@ -273,7 +273,7 @@ func TestVacuumReturnsFreePages(t *testing.T) {
|
|||||||
writeTableChanges(t, db, table)
|
writeTableChanges(t, db, table)
|
||||||
|
|
||||||
// The peer going down removes every prefix, and so every row.
|
// The peer going down removes every prefix, and so every row.
|
||||||
table.PeersDown([]string{peerA})
|
table.PeerDown(peerA)
|
||||||
writeTableChanges(t, db, table)
|
writeTableChanges(t, db, table)
|
||||||
|
|
||||||
var before int
|
var before int
|
||||||
|
|||||||
@@ -107,7 +107,7 @@ func TestReadRouteTableGivesBackWhatWasWritten(t *testing.T) {
|
|||||||
|
|
||||||
// A peer moves to another origin, one goes down and a new one comes.
|
// A peer moves to another origin, one goes down and a new one comes.
|
||||||
announceTo(source, tablePrefixV4, asB, peerA, 60)
|
announceTo(source, tablePrefixV4, asB, peerA, 60)
|
||||||
source.PeersDown([]string{peerC})
|
source.PeerDown(peerC)
|
||||||
announceTo(source, tablePrefixV6, asB, peerE, 70)
|
announceTo(source, tablePrefixV6, asB, peerE, 70)
|
||||||
|
|
||||||
writeTableChanges(t, db, source)
|
writeTableChanges(t, db, source)
|
||||||
@@ -133,7 +133,7 @@ func TestReadRouteTableGivesATableThatTakesLaterUpdates(t *testing.T) {
|
|||||||
// another one goes down, and a new peer comes.
|
// another one goes down, and a new peer comes.
|
||||||
for _, table := range []*routetable.Table{source, read} {
|
for _, table := range []*routetable.Table{source, read} {
|
||||||
table.Withdraw(netip.MustParsePrefix(tablePrefixV4), peerA)
|
table.Withdraw(netip.MustParsePrefix(tablePrefixV4), peerA)
|
||||||
table.PeersDown([]string{peerB})
|
table.PeerDown(peerB)
|
||||||
announceTo(table, tablePrefixShortV4, asB, peerD, 50)
|
announceTo(table, tablePrefixShortV4, asB, peerD, 50)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -79,9 +79,8 @@ type OriginPeers struct {
|
|||||||
|
|
||||||
// Overview works out the table's overview. It lists at most maxListed bogon
|
// Overview works out the table's overview. It lists at most maxListed bogon
|
||||||
// announcements and at most maxListed prefixes with more than one origin AS:
|
// announcements and at most maxListed prefixes with more than one origin AS:
|
||||||
// the first in prefix order. It visits every prefix in the table, in batches
|
// the first in prefix order. It walks over every prefix, but holds the read
|
||||||
// of walkBatch. After each batch, it lets the table's lock go for as long as
|
// lock for walkBatch prefixes at a time.
|
||||||
// it held the lock for the batch.
|
|
||||||
func (t *Table) Overview(maxListed int) Overview {
|
func (t *Table) Overview(maxListed int) Overview {
|
||||||
var overview Overview
|
var overview Overview
|
||||||
|
|
||||||
@@ -111,18 +110,15 @@ func (t *Table) Overview(maxListed int) Overview {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// walkPrefixes calls visit with every prefix of the table and its origins. It
|
// walkPrefixes calls visit with every prefix of the table and its origins. It
|
||||||
// holds the read lock for walkBatch prefixes at a time. After each batch, it
|
// holds the read lock, but lets it go after every walkBatch prefixes, so the
|
||||||
// lets the table's lock go for as long as it held the lock for the batch.
|
// updates waiting for it get in. A prefix that is added or removed in such a
|
||||||
// During a walk over many batches, updates can then take the lock about half
|
// gap may or may not be visited, as Go allows for a map that changes during a
|
||||||
// of the time. The walk may or may not visit a prefix that is added or removed
|
// range over it. visit must not keep origins once it returns.
|
||||||
// while the lock is let go. Go allows this in a range over a map that changes.
|
|
||||||
// visit must not keep origins once it returns.
|
|
||||||
func (t *Table) walkPrefixes(visit func(prefix netip.Prefix, origins []*origin)) {
|
func (t *Table) walkPrefixes(visit func(prefix netip.Prefix, origins []*origin)) {
|
||||||
t.mu.RLock()
|
t.mu.RLock()
|
||||||
defer t.mu.RUnlock()
|
defer t.mu.RUnlock()
|
||||||
|
|
||||||
visited := 0
|
visited := 0
|
||||||
batchStart := time.Now()
|
|
||||||
|
|
||||||
for _, f := range []*family{&t.ipv4, &t.ipv6} {
|
for _, f := range []*family{&t.ipv4, &t.ipv6} {
|
||||||
for _, prefixes := range f.byLength {
|
for _, prefixes := range f.byLength {
|
||||||
@@ -131,13 +127,8 @@ func (t *Table) walkPrefixes(visit func(prefix netip.Prefix, origins []*origin))
|
|||||||
|
|
||||||
visited++
|
visited++
|
||||||
if visited%walkBatch == 0 {
|
if visited%walkBatch == 0 {
|
||||||
held := time.Since(batchStart)
|
|
||||||
|
|
||||||
t.mu.RUnlock()
|
t.mu.RUnlock()
|
||||||
time.Sleep(held)
|
|
||||||
t.mu.RLock()
|
t.mu.RLock()
|
||||||
|
|
||||||
batchStart = time.Now()
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -131,61 +131,3 @@ func TestWalkPrefixesLetsUpdatesIn(t *testing.T) {
|
|||||||
"before its second batch")
|
"before its second batch")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestOverviewLetsUpdatesIn applies another peer's 1,000 announcements while
|
|
||||||
// Overview walks a table with more prefixes than five of the walk's batches.
|
|
||||||
// After each batch, the walk lets the lock go for as long as it held the lock.
|
|
||||||
// So all 1,000 announcements must be applied before Overview returns. A walk
|
|
||||||
// that takes the lock back at once lets in about one update per batch.
|
|
||||||
func TestOverviewLetsUpdatesIn(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
const (
|
|
||||||
asn = 1000
|
|
||||||
peer = "192.0.2.1"
|
|
||||||
otherPeer = "192.0.2.2"
|
|
||||||
// prefixCount is more prefixes than five of the walk's batches.
|
|
||||||
prefixCount = 5*walkBatch + 1
|
|
||||||
announcements = 1000
|
|
||||||
)
|
|
||||||
|
|
||||||
routeTime := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
|
||||||
table := New(Params{})
|
|
||||||
prefixes := make([]netip.Prefix, 0, prefixCount)
|
|
||||||
addr := netip.MustParseAddr("1.0.0.0")
|
|
||||||
|
|
||||||
for range prefixCount {
|
|
||||||
prefix := netip.PrefixFrom(addr, addr.BitLen())
|
|
||||||
table.Announce(prefix, asn, Route{Peer: peer, Time: routeTime})
|
|
||||||
prefixes = append(prefixes, prefix)
|
|
||||||
addr = addr.Next()
|
|
||||||
}
|
|
||||||
|
|
||||||
overviewDone := make(chan struct{})
|
|
||||||
|
|
||||||
go func() {
|
|
||||||
defer close(overviewDone)
|
|
||||||
|
|
||||||
table.Overview(0)
|
|
||||||
}()
|
|
||||||
|
|
||||||
// Wait until Overview holds the read lock. TryLock fails while a reader
|
|
||||||
// holds it.
|
|
||||||
for table.mu.TryLock() {
|
|
||||||
table.mu.Unlock()
|
|
||||||
runtime.Gosched()
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, prefix := range prefixes[:announcements] {
|
|
||||||
table.Announce(prefix, asn, Route{Peer: otherPeer, Time: routeTime})
|
|
||||||
}
|
|
||||||
|
|
||||||
select {
|
|
||||||
case <-overviewDone:
|
|
||||||
t.Fatal("Overview returned before another peer's 1,000 announcements " +
|
|
||||||
"were applied")
|
|
||||||
default:
|
|
||||||
}
|
|
||||||
|
|
||||||
<-overviewDone
|
|
||||||
}
|
|
||||||
@@ -227,7 +227,7 @@ func TestCounts(t *testing.T) {
|
|||||||
})
|
})
|
||||||
|
|
||||||
// peerC was the last peer of the IPv6 prefix. It keeps its number.
|
// peerC was the last peer of the IPv6 prefix. It keeps its number.
|
||||||
table.PeersDown([]string{peerC})
|
table.PeerDown(peerC)
|
||||||
checkCounts(t, table, routetable.Counts{
|
checkCounts(t, table, routetable.Counts{
|
||||||
Pairs: 2,
|
Pairs: 2,
|
||||||
Peers: 3,
|
Peers: 3,
|
||||||
|
|||||||
@@ -28,7 +28,7 @@ const (
|
|||||||
keptTimesDegree = 32
|
keptTimesDegree = 32
|
||||||
|
|
||||||
// walkBatch is how many prefixes a walk over the whole table visits in
|
// walkBatch is how many prefixes a walk over the whole table visits in
|
||||||
// one hold of the lock. Overview and PeersDown each walk the whole table.
|
// one hold of the lock. Overview and PeerDown each walk the whole table.
|
||||||
// Callers waiting for the lock get it between batches.
|
// Callers waiting for the lock get it between batches.
|
||||||
walkBatch = 10000
|
walkBatch = 10000
|
||||||
|
|
||||||
@@ -205,29 +205,20 @@ func (t *Table) Withdraw(prefix netip.Prefix, peer string) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// PeersDown applies the sessions of peers going down: each of the peers leaves
|
// PeerDown applies a peer's session going down: the peer leaves every peer
|
||||||
// every peer set, and every route of each is dropped. When any of them carries
|
// set, and every route of the peer is dropped. When the peer carries any
|
||||||
// a prefix, PeersDown visits every prefix in the table once, in batches of
|
// prefix, PeerDown visits every prefix in the table, in batches of walkBatch.
|
||||||
// walkBatch. After each batch, it lets the table's lock go for as long as it
|
// After each batch, it lets the table's lock go for as long as it held the
|
||||||
// held the lock for the batch. During a walk over many batches, reads and
|
// lock for the batch. During a walk over many batches, reads and updates can
|
||||||
// updates can then take the lock about half of the time. An announcement by
|
// then take the lock about half of the time. An announcement by the peer
|
||||||
// one of the peers applied during the walk may be dropped too. So the caller
|
// applied during the walk may be dropped too. So the caller applies the peer's
|
||||||
// applies those peers' later updates only once PeersDown has returned.
|
// later updates only once PeerDown has returned.
|
||||||
func (t *Table) PeersDown(peers []string) {
|
func (t *Table) PeerDown(peer string) {
|
||||||
t.mu.Lock()
|
t.mu.Lock()
|
||||||
defer t.mu.Unlock()
|
defer t.mu.Unlock()
|
||||||
|
|
||||||
// numbers holds the number of each of the peers that carries a prefix.
|
number, known := t.peerNumbers[peer]
|
||||||
numbers := make([]uint, 0, len(peers))
|
if !known || t.peerPrefixes[number] == 0 {
|
||||||
|
|
||||||
for _, peer := range peers {
|
|
||||||
number, known := t.peerNumbers[peer]
|
|
||||||
if known && t.peerPrefixes[number] > 0 {
|
|
||||||
numbers = append(numbers, number)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if len(numbers) == 0 {
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -237,15 +228,9 @@ func (t *Table) PeersDown(peers []string) {
|
|||||||
for _, f := range []*family{&t.ipv4, &t.ipv6} {
|
for _, f := range []*family{&t.ipv4, &t.ipv6} {
|
||||||
for _, prefixes := range f.byLength {
|
for _, prefixes := range f.byLength {
|
||||||
for prefix, origins := range prefixes {
|
for prefix, origins := range prefixes {
|
||||||
for _, number := range numbers {
|
o := originWithPeer(origins, number)
|
||||||
o := originWithPeer(origins, number)
|
if o != nil {
|
||||||
if o != nil {
|
t.leave(prefix, o, number)
|
||||||
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
|
// Go allows the map to change before the next step of
|
||||||
|
|||||||
@@ -7,11 +7,11 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
// TestPeersDownLetsReadsIn reads the table's counts again and again while a
|
// TestPeerDownLetsReadsIn reads the table's counts again and again while a peer
|
||||||
// peer that carries more than two batches of prefixes goes down. PeersDown lets
|
// that carries more than two batches of prefixes goes down. PeerDown lets the
|
||||||
// the lock go after every walkBatch prefixes, so a read must get in while the
|
// lock go after every walkBatch prefixes, so a read must get in while the peer
|
||||||
// peer still carries some of its prefixes but no longer all.
|
// still carries some of its prefixes but no longer all.
|
||||||
func TestPeersDownLetsReadsIn(t *testing.T) {
|
func TestPeerDownLetsReadsIn(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
const (
|
const (
|
||||||
@@ -32,8 +32,8 @@ func TestPeersDownLetsReadsIn(t *testing.T) {
|
|||||||
addr = addr.Next()
|
addr = addr.Next()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Hold a read lock until PeersDown waits for the lock, so that the first
|
// Hold a read lock until PeerDown waits for the lock, so that the first
|
||||||
// read below starts once PeersDown holds it. TryRLock fails once a writer
|
// read below starts once PeerDown holds it. TryRLock fails once a writer
|
||||||
// waits.
|
// waits.
|
||||||
table.mu.RLock()
|
table.mu.RLock()
|
||||||
|
|
||||||
@@ -42,7 +42,7 @@ func TestPeersDownLetsReadsIn(t *testing.T) {
|
|||||||
go func() {
|
go func() {
|
||||||
defer close(peerDownDone)
|
defer close(peerDownDone)
|
||||||
|
|
||||||
table.PeersDown([]string{peer})
|
table.PeerDown(peer)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
for table.mu.TryRLock() {
|
for table.mu.TryRLock() {
|
||||||
@@ -60,7 +60,7 @@ func TestPeersDownLetsReadsIn(t *testing.T) {
|
|||||||
|
|
||||||
select {
|
select {
|
||||||
case <-peerDownDone:
|
case <-peerDownDone:
|
||||||
t.Fatal("no read got in while PeersDown ran")
|
t.Fatal("no read got in while PeerDown ran")
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -69,10 +69,10 @@ func withdraw(peer string) step {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// peersDown is the step of the sessions of peers going down.
|
// peerDown is the step of peer's session going down.
|
||||||
func peersDown(peers ...string) step {
|
func peerDown(peer string) step {
|
||||||
return func(table *routetable.Table, _ netip.Prefix) {
|
return func(table *routetable.Table, _ netip.Prefix) {
|
||||||
table.PeersDown(peers)
|
table.PeerDown(peer)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -272,28 +272,21 @@ func TestWithdrawalsAndPeerDowns(t *testing.T) {
|
|||||||
{
|
{
|
||||||
name: "a peer going down leaves the prefix",
|
name: "a peer going down leaves the prefix",
|
||||||
steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
|
steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
|
||||||
peersDown(peerB)},
|
peerDown(peerB)},
|
||||||
want: []routetable.Origin{{ASN: asA, Peers: []string{peerA},
|
want: []routetable.Origin{{ASN: asA, Peers: []string{peerA},
|
||||||
Routes: []routetable.Route{routeOf(peerA, asA, 10)}}},
|
Routes: []routetable.Route{routeOf(peerA, asA, 10)}}},
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
name: "a peer going down after it withdrew changes nothing",
|
name: "a peer going down after it withdrew changes nothing",
|
||||||
steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
|
steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
|
||||||
withdraw(peerB), peersDown(peerB)},
|
withdraw(peerB), peerDown(peerB)},
|
||||||
want: []routetable.Origin{{ASN: asA, Peers: []string{peerA},
|
want: []routetable.Origin{{ASN: asA, Peers: []string{peerA},
|
||||||
Routes: []routetable.Route{routeOf(peerA, asA, 10)}}},
|
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 TestPeersDownLeavesEveryPrefix(t *testing.T) {
|
func TestPeerDownLeavesEveryPrefix(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
table := routetable.New(routetable.Params{})
|
table := routetable.New(routetable.Params{})
|
||||||
@@ -310,7 +303,7 @@ func TestPeersDownLeavesEveryPrefix(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
table.Announce(prefixes[0], asB, routeOf(peerB, asB, 20))
|
table.Announce(prefixes[0], asB, routeOf(peerB, asB, 20))
|
||||||
table.PeersDown([]string{peerA})
|
table.PeerDown(peerA)
|
||||||
|
|
||||||
want := []routetable.Origin{{ASN: asB, Peers: []string{peerB},
|
want := []routetable.Origin{{ASN: asB, Peers: []string{peerB},
|
||||||
Routes: []routetable.Route{routeOf(peerB, asB, 20)}}}
|
Routes: []routetable.Route{routeOf(peerB, asB, 20)}}}
|
||||||
@@ -340,32 +333,6 @@ func TestPeersDownLeavesEveryPrefix(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) {
|
func TestCountsOriginsWithoutKeptRoute(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
@@ -391,7 +358,7 @@ func TestCountsOriginsWithoutKeptRoute(t *testing.T) {
|
|||||||
{announce(peerC, asA, 30), 0},
|
{announce(peerC, asA, 30), 0},
|
||||||
{withdraw(peerB), 0},
|
{withdraw(peerB), 0},
|
||||||
{withdraw(peerC), 1},
|
{withdraw(peerC), 1},
|
||||||
{peersDown(peerA), 0},
|
{peerDown(peerA), 0},
|
||||||
}
|
}
|
||||||
|
|
||||||
for i, s := range steps {
|
for i, s := range steps {
|
||||||
@@ -497,7 +464,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
|
|||||||
checkChanges(t, table, []change{}, map[string]uint{})
|
checkChanges(t, table, []change{}, map[string]uint{})
|
||||||
|
|
||||||
// peerA going down changes v4, the one prefix it carries.
|
// peerA going down changes v4, the one prefix it carries.
|
||||||
table.PeersDown([]string{peerA})
|
table.PeerDown(peerA)
|
||||||
checkChanges(t, table, []change{
|
checkChanges(t, table, []change{
|
||||||
{Prefix: prefixV4, OriginASN: asA, Peers: []uint{1, 2},
|
{Prefix: prefixV4, OriginASN: asA, Peers: []uint{1, 2},
|
||||||
Routes: []routetable.Route{
|
Routes: []routetable.Route{
|
||||||
@@ -506,7 +473,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
|
|||||||
}, map[string]uint{})
|
}, map[string]uint{})
|
||||||
|
|
||||||
// peerA, which carries nothing now, goes down again.
|
// peerA, which carries nothing now, goes down again.
|
||||||
table.PeersDown([]string{peerA})
|
table.PeerDown(peerA)
|
||||||
checkChanges(t, table, []change{}, map[string]uint{})
|
checkChanges(t, table, []change{}, map[string]uint{})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -650,7 +617,7 @@ func TestConcurrentUse(t *testing.T) {
|
|||||||
func(round int) { table.Announce(prefix, asA, routeOf(peerA, asA, round)) },
|
func(round int) { table.Announce(prefix, asA, routeOf(peerA, asA, round)) },
|
||||||
func(round int) { table.Announce(prefix, asB, routeOf(peerB, asB, round)) },
|
func(round int) { table.Announce(prefix, asB, routeOf(peerB, asB, round)) },
|
||||||
func(int) { table.Withdraw(prefix, peerA) },
|
func(int) { table.Withdraw(prefix, peerA) },
|
||||||
func(int) { table.PeersDown([]string{peerB}) },
|
func(int) { table.PeerDown(peerB) },
|
||||||
func(int) { table.PutBackChanges(table.TakeChanges(math.MaxInt)) },
|
func(int) { table.PutBackChanges(table.TakeChanges(math.MaxInt)) },
|
||||||
func(int) { table.ChangeCount() },
|
func(int) { table.ChangeCount() },
|
||||||
func(int) { table.Lookup(prefix.Addr()) },
|
func(int) { table.Lookup(prefix.Addr()) },
|
||||||
|
|||||||
@@ -46,19 +46,10 @@ const (
|
|||||||
maxTableChangesPerWrite = 20000
|
maxTableChangesPerWrite = 20000
|
||||||
|
|
||||||
// maxHeldBackMessages is the most messages the handler holds back from
|
// maxHeldBackMessages is the most messages the handler holds back from
|
||||||
// peers whose session went down. A peer's messages are held back from
|
// peers whose PeerDown has not returned. A message takes about 1 KiB, so
|
||||||
// the moment its session goes down until peerDown has applied them. If
|
// that many take about 20 MiB (README "Memory"). At this number
|
||||||
// peerDown is running at that moment, the peer waits for the rest of
|
// HandleMessage waits until one of them is applied.
|
||||||
// peerDown's current PeersDown call. The peer also waits while peerDown
|
maxHeldBackMessages = 20000
|
||||||
// applies the messages held back from that call's peers, those that
|
|
||||||
// arrive meanwhile included. The next PeersDown call takes the peer out.
|
|
||||||
// Once that call returns, peerDown applies the messages held back from
|
|
||||||
// that call's peers, one peer after another, in the order the call lists
|
|
||||||
// them. A held-back message that says the peer's session went down again
|
|
||||||
// makes the peer's later messages wait for the next call. A message takes
|
|
||||||
// about 1 KiB, so 150,000 take about 150 MiB (README "Memory"). At this
|
|
||||||
// number HandleMessage waits until one of them is applied.
|
|
||||||
maxHeldBackMessages = 150000
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// PrefixHandler keeps the routing table in memory current from the feed. It
|
// PrefixHandler keeps the routing table in memory current from the feed. It
|
||||||
@@ -67,16 +58,14 @@ const (
|
|||||||
// does an announcement whose AS path is empty or holds AS 0 anywhere. A peer
|
// does an announcement whose AS path is empty or holds AS 0 anywhere. A peer
|
||||||
// whose BGP session goes down leaves every prefix.
|
// whose BGP session goes down leaves every prefix.
|
||||||
//
|
//
|
||||||
// The table's PeersDown applies sessions going down. It visits every prefix,
|
// The table's PeerDown applies a session going down. It visits every prefix,
|
||||||
// which takes seconds on a full table. So the handler runs PeersDown on a
|
// which takes seconds on a full table. So the handler runs PeerDown on
|
||||||
// separate goroutine and goes on taking messages meanwhile. One PeersDown runs
|
// a separate goroutine, one peer at a time, and goes on taking messages
|
||||||
// at a time. The next one takes every peer whose session went down since the
|
// meanwhile. PeerDown lets the table's lock go about half of the time, so the
|
||||||
// previous one began. PeersDown lets the table's lock go about half of the
|
// handler applies other peers' messages while PeerDown runs. The handler holds
|
||||||
// time, so the handler applies other peers' messages while PeersDown runs. The
|
// back the down peer's later messages until PeerDown returns. Then the handler
|
||||||
// handler holds back a down peer's later messages until peerDown applies them,
|
// applies them in order, one at a time, and takes new messages in between. It
|
||||||
// as the doc of maxHeldBackMessages describes. It applies them in order, one at
|
// holds back at most maxHeldBackMessages messages.
|
||||||
// a time, and takes new messages in between. It holds back at most
|
|
||||||
// maxHeldBackMessages messages.
|
|
||||||
//
|
//
|
||||||
// Each flush writes the table's changes to the database. It also stores each
|
// Each flush writes the table's changes to the database. It also stores each
|
||||||
// prefix the updates name with the time it was first seen.
|
// prefix the updates name with the time it was first seen.
|
||||||
@@ -114,25 +103,20 @@ type PrefixHandler struct {
|
|||||||
stopCh chan context.Context
|
stopCh chan context.Context
|
||||||
done chan struct{} // closed when the flush goroutine returns
|
done chan struct{} // closed when the flush goroutine returns
|
||||||
|
|
||||||
// Peers whose session went down. heldBack, under mu, holds each such
|
// Peers whose session went down. peerDownMu lets one PeerDown run at a
|
||||||
// peer's later messages, in order, until the peer has left every prefix
|
// time, so the handler's own updates wait for the table's lock behind
|
||||||
// and they are applied. A peer with no such messages yet has an empty
|
// at most one. heldBack, under mu, holds each such peer's later
|
||||||
// entry. heldCount, under mu, is how many messages heldBack holds.
|
// messages, in order, until its PeerDown has returned and they are
|
||||||
|
// applied. A peer with no such messages yet has an empty entry.
|
||||||
|
// heldCount, under mu, is how many messages heldBack holds.
|
||||||
// heldTaken is signalled each time a held-back message is taken out to
|
// heldTaken is signalled each time a held-back message is taken out to
|
||||||
// be applied; its L is &mu.
|
// be applied; its L is &mu. peerDowns counts the peerDown goroutines
|
||||||
heldBack map[string][]*ristypes.RISMessage
|
// that have not returned.
|
||||||
heldCount int
|
peerDownMu sync.Mutex
|
||||||
heldTaken sync.Cond
|
heldBack map[string][]*ristypes.RISMessage
|
||||||
// downPeers, under mu, lists the peers whose session went down and that
|
heldCount int
|
||||||
// the next PeersDown takes. peerDownRunning, under mu, says whether
|
heldTaken sync.Cond
|
||||||
// peerDown is running. peerDowns counts the peerDown goroutines that have
|
peerDowns sync.WaitGroup
|
||||||
// not returned.
|
|
||||||
downPeers []string
|
|
||||||
peerDownRunning bool
|
|
||||||
peerDowns sync.WaitGroup
|
|
||||||
// tablePeersDown is the table's PeersDown. Only peerDown calls it, so a
|
|
||||||
// test may replace it before it hands the handler a message.
|
|
||||||
tablePeersDown func(peers []string)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// seenPrefix is what the batch keeps of an update: the prefix it names and its
|
// seenPrefix is what the batch keeps of an update: the prefix it names and its
|
||||||
@@ -197,7 +181,6 @@ func newPrefixHandler(
|
|||||||
stopCh: make(chan context.Context, 1),
|
stopCh: make(chan context.Context, 1),
|
||||||
done: make(chan struct{}),
|
done: make(chan struct{}),
|
||||||
heldBack: make(map[string][]*ristypes.RISMessage),
|
heldBack: make(map[string][]*ristypes.RISMessage),
|
||||||
tablePeersDown: table.PeersDown,
|
|
||||||
}
|
}
|
||||||
h.batchTaken.L = &h.mu
|
h.batchTaken.L = &h.mu
|
||||||
h.heldTaken.L = &h.mu
|
h.heldTaken.L = &h.mu
|
||||||
@@ -229,7 +212,7 @@ func (h *PrefixHandler) QueueCapacity() int {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// HandleMessage processes a message to track prefix information. A message
|
// HandleMessage processes a message to track prefix information. A message
|
||||||
// from a peer whose session went down is held back until peerDown applies it.
|
// from a peer whose PeerDown has not returned is held back until it has.
|
||||||
func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
|
func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
|
||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
defer h.mu.Unlock()
|
defer h.mu.Unlock()
|
||||||
@@ -252,16 +235,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
|
|||||||
h.applyMessage(msg)
|
h.applyMessage(msg)
|
||||||
}
|
}
|
||||||
|
|
||||||
// HeldBackCount returns how many messages the handler holds back from peers
|
// Stop waits for every PeerDown the handler started and for the messages held
|
||||||
// whose session went down. The status page shows it.
|
|
||||||
func (h *PrefixHandler) HeldBackCount() int {
|
|
||||||
h.mu.Lock()
|
|
||||||
defer h.mu.Unlock()
|
|
||||||
|
|
||||||
return h.heldCount
|
|
||||||
}
|
|
||||||
|
|
||||||
// Stop waits for every PeersDown the handler started and for the messages held
|
|
||||||
// back meanwhile to be applied. It gives up this wait once ctx ends. Then Stop
|
// back meanwhile to be applied. It gives up this wait once ctx ends. Then Stop
|
||||||
// stops the flush goroutine and waits for it until ctx ends. The flush
|
// stops the flush goroutine and waits for it until ctx ends. The flush
|
||||||
// goroutine makes one last flush before it returns. That flush writes at most
|
// goroutine makes one last flush before it returns. That flush writes at most
|
||||||
@@ -316,8 +290,8 @@ func (h *PrefixHandler) Stop(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// applyMessage applies msg. The caller holds mu. A peer's session going down
|
// applyMessage applies msg. The caller holds mu. A peer's session going down
|
||||||
// has the peer leave every prefix. An update is applied to the routing table,
|
// starts the peer's PeerDown. An update is applied to the routing table, and
|
||||||
// and its prefixes join the batch.
|
// its prefixes join the batch.
|
||||||
func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
|
func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
|
||||||
if msg.Type == ristypes.MessageTypeState {
|
if msg.Type == ristypes.MessageTypeState {
|
||||||
if msg.State == ristypes.StateDown {
|
if msg.State == ristypes.StateDown {
|
||||||
@@ -396,64 +370,34 @@ func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// startPeerDown adds peer to downPeers. It starts peerDown on a new goroutine
|
// startPeerDown starts peerDown for peer on a new goroutine. The caller holds
|
||||||
// unless peerDown is running. The caller holds mu. The peer's messages are
|
// mu. The peer's messages are held back until peerDown has applied them.
|
||||||
// held back until peerDown has applied them. Messages held back for an earlier
|
// Messages held back for an earlier PeerDown of the peer, and not yet applied,
|
||||||
// session of the peer going down, and not yet applied, stay held back.
|
// stay held back.
|
||||||
func (h *PrefixHandler) startPeerDown(peer string) {
|
func (h *PrefixHandler) startPeerDown(peer string) {
|
||||||
_, goingDown := h.heldBack[peer]
|
_, goingDown := h.heldBack[peer]
|
||||||
if !goingDown {
|
if !goingDown {
|
||||||
h.heldBack[peer] = []*ristypes.RISMessage{}
|
h.heldBack[peer] = []*ristypes.RISMessage{}
|
||||||
}
|
}
|
||||||
|
|
||||||
h.downPeers = append(h.downPeers, peer)
|
|
||||||
|
|
||||||
if h.peerDownRunning {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
h.peerDownRunning = true
|
|
||||||
h.peerDowns.Add(1)
|
h.peerDowns.Add(1)
|
||||||
|
|
||||||
go h.peerDown()
|
go h.peerDown(peer)
|
||||||
}
|
}
|
||||||
|
|
||||||
// peerDown takes every peer in downPeers and runs the table's PeersDown for
|
// peerDown runs the table's PeerDown for peer, once no other PeerDown is
|
||||||
// them. Then it applies the messages held back from each of those peers. It
|
// running. Then peerDown applies the messages held back from the peer, in
|
||||||
// repeats this until downPeers is empty.
|
// order. It takes mu for one message at a time, so HandleMessage takes new
|
||||||
func (h *PrefixHandler) peerDown() {
|
// messages in between. The peer's new messages join the end of the held-back
|
||||||
|
// ones. A held-back message that says the peer's session went down again
|
||||||
|
// starts a new peerDown, which applies the rest.
|
||||||
|
func (h *PrefixHandler) peerDown(peer string) {
|
||||||
defer h.peerDowns.Done()
|
defer h.peerDowns.Done()
|
||||||
|
|
||||||
for {
|
h.peerDownMu.Lock()
|
||||||
h.mu.Lock()
|
h.table.PeerDown(peer)
|
||||||
|
h.peerDownMu.Unlock()
|
||||||
|
|
||||||
peers := h.downPeers
|
|
||||||
h.downPeers = nil
|
|
||||||
|
|
||||||
if len(peers) == 0 {
|
|
||||||
h.peerDownRunning = false
|
|
||||||
h.mu.Unlock()
|
|
||||||
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
h.mu.Unlock()
|
|
||||||
|
|
||||||
h.tablePeersDown(peers)
|
|
||||||
|
|
||||||
for _, peer := range peers {
|
|
||||||
h.applyHeldBack(peer)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// applyHeldBack applies the messages held back from peer, in order. It takes
|
|
||||||
// mu for one message at a time, so HandleMessage takes new messages in
|
|
||||||
// between. The peer's new messages join the end of the held-back ones. A
|
|
||||||
// held-back message that says the peer's session went down again adds the
|
|
||||||
// peer to downPeers. The rest stay held back until the next PeersDown has
|
|
||||||
// returned.
|
|
||||||
func (h *PrefixHandler) applyHeldBack(peer string) {
|
|
||||||
for {
|
for {
|
||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
|
|
||||||
|
|||||||
@@ -238,7 +238,7 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// TestPrefixHandlerStopLogsTheMessagesHeldBack stops a prefix handler with a
|
// TestPrefixHandlerStopLogsTheMessagesHeldBack stops a prefix handler with a
|
||||||
// deadline that has passed, while a peer's PeersDown is held up and one message
|
// deadline that has passed, while a peer's PeerDown is held up and one message
|
||||||
// of the peer is held back. Stop must log that message as not applied.
|
// of the peer is held back. Stop must log that message as not applied.
|
||||||
func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
|
func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
@@ -249,8 +249,8 @@ func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
|
|||||||
routetable.New(routetable.Params{}), longerThanAnyTest,
|
routetable.New(routetable.Params{}), longerThanAnyTest,
|
||||||
maxTableChangesPerWrite)
|
maxTableChangesPerWrite)
|
||||||
|
|
||||||
release := holdUpPeersDown(h)
|
h.peerDownMu.Lock()
|
||||||
defer close(release)
|
defer h.peerDownMu.Unlock()
|
||||||
|
|
||||||
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
|
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
|
||||||
h.HandleMessage(sessionDown())
|
h.HandleMessage(sessionDown())
|
||||||
@@ -268,8 +268,8 @@ func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// TestPrefixHandlerStopWaitsForTheMessagesHeldBack stops a prefix handler
|
// TestPrefixHandlerStopWaitsForTheMessagesHeldBack stops a prefix handler
|
||||||
// while a peer's PeersDown is held up and an announcement of the peer is held
|
// while a peer's PeerDown is held up and an announcement of the peer is held
|
||||||
// back. Stop must wait. Once the PeersDown is let go and Stop returns, the
|
// back. Stop must wait. Once the PeerDown is let go and Stop returns, the
|
||||||
// announcement must be in the database.
|
// announcement must be in the database.
|
||||||
func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
|
func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
@@ -278,7 +278,7 @@ func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
|
|||||||
h := newPrefixHandler(db, logger.New(), routetable.New(routetable.Params{}),
|
h := newPrefixHandler(db, logger.New(), routetable.New(routetable.Params{}),
|
||||||
longerThanAnyTest, maxTableChangesPerWrite)
|
longerThanAnyTest, maxTableChangesPerWrite)
|
||||||
|
|
||||||
release := holdUpPeersDown(h)
|
h.peerDownMu.Lock()
|
||||||
h.HandleMessage(sessionDown())
|
h.HandleMessage(sessionDown())
|
||||||
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
|
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
|
||||||
|
|
||||||
@@ -291,16 +291,16 @@ func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
|
|||||||
|
|
||||||
select {
|
select {
|
||||||
case <-stopped:
|
case <-stopped:
|
||||||
t.Error("Stop returned while a PeersDown was held up")
|
t.Error("Stop returned while a PeerDown was held up")
|
||||||
case <-time.After(stillWaitingAfter):
|
case <-time.After(stillWaitingAfter):
|
||||||
}
|
}
|
||||||
|
|
||||||
close(release)
|
h.peerDownMu.Unlock()
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-stopped:
|
case <-stopped:
|
||||||
case <-time.After(stepTimeout):
|
case <-time.After(stepTimeout):
|
||||||
t.Fatalf("Stop did not return within %s of the PeersDown's release",
|
t.Fatalf("Stop did not return within %s of the PeerDown's release",
|
||||||
stepTimeout)
|
stepTimeout)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -324,32 +324,16 @@ const longerThanAnyTest = time.Hour
|
|||||||
// applies the updates to an empty routing table. It is built without
|
// applies the updates to an empty routing table. It is built without
|
||||||
// NewPrefixHandler's flush goroutine, so the test flushes itself.
|
// NewPrefixHandler's flush goroutine, so the test flushes itself.
|
||||||
func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
|
func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
|
||||||
table := routetable.New(routetable.Params{})
|
|
||||||
|
|
||||||
return &PrefixHandler{
|
return &PrefixHandler{
|
||||||
db: store,
|
db: store,
|
||||||
logger: logger.New(),
|
logger: logger.New(),
|
||||||
table: table,
|
table: routetable.New(routetable.Params{}),
|
||||||
batchMaxSize: prefixBatchMaxSize,
|
batchMaxSize: prefixBatchMaxSize,
|
||||||
tableChangesPerWrite: maxTableChangesPerWrite,
|
tableChangesPerWrite: maxTableChangesPerWrite,
|
||||||
heldBack: make(map[string][]*ristypes.RISMessage),
|
heldBack: make(map[string][]*ristypes.RISMessage),
|
||||||
tablePeersDown: table.PeersDown,
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// holdUpPeersDown makes each PeersDown of h wait until the returned channel is
|
|
||||||
// closed, as another peer's long PeersDown would hold it up. The test calls it
|
|
||||||
// before it hands h a message.
|
|
||||||
func holdUpPeersDown(h *PrefixHandler) chan struct{} {
|
|
||||||
release := make(chan struct{})
|
|
||||||
h.tablePeersDown = func(peers []string) {
|
|
||||||
<-release
|
|
||||||
h.table.PeersDown(peers)
|
|
||||||
}
|
|
||||||
|
|
||||||
return release
|
|
||||||
}
|
|
||||||
|
|
||||||
// newTestDatabase opens a database in a new temporary directory. It is closed
|
// newTestDatabase opens a database in a new temporary directory. It is closed
|
||||||
// when the test ends.
|
// when the test ends.
|
||||||
func newTestDatabase(t *testing.T) *database.Database {
|
func newTestDatabase(t *testing.T) *database.Database {
|
||||||
|
|||||||
@@ -103,25 +103,24 @@ func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeersDown of a
|
// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeerDown of a
|
||||||
// peer whose session went down, as another peer's long PeersDown would. The
|
// peer whose session went down, as another peer's long PeerDown would. The
|
||||||
// handler must go on taking messages. It applies another peer's announcement
|
// handler must go on taking messages. It applies another peer's announcement
|
||||||
// at once. It holds back these messages of the down peer, in this order:
|
// at once. It holds back these messages of the down peer, in this order:
|
||||||
// - an announcement;
|
// - an announcement;
|
||||||
// - its session going down again;
|
// - its session going down again;
|
||||||
// - a second announcement.
|
// - a second announcement.
|
||||||
//
|
//
|
||||||
// HeldBackCount must count them. Once the PeersDown calls have returned, the
|
// Once the PeerDowns have returned, the down peer must carry only the prefix
|
||||||
// down peer must carry only the prefix of the second announcement.
|
// of the second announcement.
|
||||||
func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
|
func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
const heldMessages = 3
|
|
||||||
|
|
||||||
h := prefixHandlerWithoutTimer(newMockStore())
|
h := prefixHandlerWithoutTimer(newMockStore())
|
||||||
release := holdUpPeersDown(h)
|
|
||||||
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
|
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
|
||||||
|
|
||||||
|
h.peerDownMu.Lock()
|
||||||
|
|
||||||
handled := returnsInTime(func() {
|
handled := returnsInTime(func() {
|
||||||
for _, msg := range []*ristypes.RISMessage{
|
for _, msg := range []*ristypes.RISMessage{
|
||||||
sessionDown(),
|
sessionDown(),
|
||||||
@@ -145,15 +144,10 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
|
|||||||
setV4Prefix: {},
|
setV4Prefix: {},
|
||||||
})
|
})
|
||||||
|
|
||||||
if h.HeldBackCount() != heldMessages {
|
h.peerDownMu.Unlock()
|
||||||
t.Errorf("HeldBackCount() = %d, want %d", h.HeldBackCount(),
|
|
||||||
heldMessages)
|
|
||||||
}
|
|
||||||
|
|
||||||
close(release)
|
|
||||||
|
|
||||||
if !returnsInTime(h.peerDowns.Wait) {
|
if !returnsInTime(h.peerDowns.Wait) {
|
||||||
t.Fatalf("the PeersDown calls did not return within %s", stepTimeout)
|
t.Fatalf("the PeerDowns did not return within %s", stepTimeout)
|
||||||
}
|
}
|
||||||
|
|
||||||
checkTableOrigins(t, h.table, map[string]map[int][]string{
|
checkTableOrigins(t, h.table, map[string]map[int][]string{
|
||||||
@@ -164,85 +158,21 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestPrefixHandlerTakesPeersDownTogether has four peers announce testV4Prefix.
|
// TestPrefixHandlerAppliesMessagesWhilePeerDownRuns gives a peer 50,000
|
||||||
// It holds up the PeersDown of the first peer, whose session goes down.
|
|
||||||
// Meanwhile the sessions of the three others go down, and each of them
|
|
||||||
// announces otherV4Prefix. Once the first PeersDown is let go, the next one
|
|
||||||
// must take the three others together. No peer may then carry testV4Prefix,
|
|
||||||
// and the three others must carry otherV4Prefix.
|
|
||||||
func TestPrefixHandlerTakesPeersDownTogether(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
others := []string{secondPeerIP, thirdPeerIP, fourthPeerIP}
|
|
||||||
h := prefixHandlerWithoutTimer(newMockStore())
|
|
||||||
|
|
||||||
// calls receives the peers of each PeersDown, with room for one call
|
|
||||||
// for each peer.
|
|
||||||
calls := make(chan []string, len(others)+1)
|
|
||||||
release := make(chan struct{})
|
|
||||||
h.tablePeersDown = func(peers []string) {
|
|
||||||
calls <- slices.Clone(peers)
|
|
||||||
|
|
||||||
<-release
|
|
||||||
h.table.PeersDown(peers)
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, peer := range append([]string{testPeerIP}, others...) {
|
|
||||||
h.HandleMessage(peerAnnouncement(peer, testV4Prefix))
|
|
||||||
}
|
|
||||||
|
|
||||||
h.HandleMessage(sessionDown())
|
|
||||||
|
|
||||||
select {
|
|
||||||
case <-calls:
|
|
||||||
case <-time.After(stepTimeout):
|
|
||||||
t.Fatalf("PeersDown did not begin within %s", stepTimeout)
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, peer := range others {
|
|
||||||
h.HandleMessage(sessionDownOf(peer))
|
|
||||||
h.HandleMessage(peerAnnouncement(peer, otherV4Prefix))
|
|
||||||
}
|
|
||||||
|
|
||||||
close(release)
|
|
||||||
|
|
||||||
if !returnsInTime(h.peerDowns.Wait) {
|
|
||||||
t.Fatalf("the PeersDown calls did not return within %s", stepTimeout)
|
|
||||||
}
|
|
||||||
|
|
||||||
close(calls)
|
|
||||||
|
|
||||||
later := make([][]string, 0, len(calls))
|
|
||||||
for peers := range calls {
|
|
||||||
later = append(later, peers)
|
|
||||||
}
|
|
||||||
|
|
||||||
if !reflect.DeepEqual(later, [][]string{others}) {
|
|
||||||
t.Errorf("the PeersDown calls after the first took %v, want %v "+
|
|
||||||
"together", later, others)
|
|
||||||
}
|
|
||||||
|
|
||||||
checkTableOrigins(t, h.table, map[string]map[int][]string{
|
|
||||||
testV4Prefix: {},
|
|
||||||
otherV4Prefix: {testASNB: others},
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPrefixHandlerAppliesMessagesWhilePeersDownRuns gives a peer 50,000
|
|
||||||
// prefixes and hands the prefix handler that peer's session going down. The
|
// prefixes and hands the prefix handler that peer's session going down. The
|
||||||
// table's PeersDown then visits the prefixes in five batches. Once PeersDown
|
// table's PeerDown then visits the prefixes in five batches. Once PeerDown has
|
||||||
// has begun, the handler is handed 1,000 announcements of another peer. The
|
// begun, the handler is handed 1,000 announcements of another peer. The
|
||||||
// handler must apply them all before PeersDown ends, while the down peer still
|
// handler must apply them all before PeerDown ends, while the down peer still
|
||||||
// carries some of its prefixes.
|
// carries some of its prefixes.
|
||||||
func TestPrefixHandlerAppliesMessagesWhilePeersDownRuns(t *testing.T) {
|
func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
const (
|
const (
|
||||||
// downPeerPrefixes is five times the 10,000 prefixes PeersDown visits
|
// downPeerPrefixes is five times the 10,000 prefixes PeerDown visits
|
||||||
// in one hold of the table's lock.
|
// in one hold of the table's lock.
|
||||||
downPeerPrefixes = 50000
|
downPeerPrefixes = 50000
|
||||||
// otherPeerMessages is far more messages than the handler applies
|
// otherPeerMessages is far more messages than the handler applies
|
||||||
// while PeersDown runs, if PeersDown takes the lock back at once after
|
// while PeerDown runs, if PeerDown takes the lock back at once after
|
||||||
// each batch. The handler then applies only a few for each batch.
|
// each batch. The handler then applies only a few for each batch.
|
||||||
otherPeerMessages = 1000
|
otherPeerMessages = 1000
|
||||||
// firstOtherAddr is the address of the other peer's first prefix.
|
// firstOtherAddr is the address of the other peer's first prefix.
|
||||||
@@ -254,7 +184,7 @@ func TestPrefixHandlerAppliesMessagesWhilePeersDownRuns(t *testing.T) {
|
|||||||
|
|
||||||
h.HandleMessage(sessionDown())
|
h.HandleMessage(sessionDown())
|
||||||
|
|
||||||
// Wait until PeersDown has begun: the down peer has left some of its
|
// Wait until PeerDown has begun: the down peer has left some of its
|
||||||
// prefixes. The down peer's prefixes are IPv4 and the other peer's are
|
// prefixes. The down peer's prefixes are IPv4 and the other peer's are
|
||||||
// IPv6, so the counts tell them apart.
|
// IPv6, so the counts tell them apart.
|
||||||
for h.table.Counts().IPv4Routes == downPeerPrefixes {
|
for h.table.Counts().IPv4Routes == downPeerPrefixes {
|
||||||
@@ -277,76 +207,15 @@ func TestPrefixHandlerAppliesMessagesWhilePeersDownRuns(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if counts.IPv4Routes == 0 {
|
if counts.IPv4Routes == 0 {
|
||||||
t.Errorf("PeersDown ended before the handler applied the other peer's "+
|
t.Errorf("PeerDown ended before the handler applied the other peer's "+
|
||||||
"%d announcements", otherPeerMessages)
|
"%d announcements", otherPeerMessages)
|
||||||
}
|
}
|
||||||
|
|
||||||
if !returnsInTime(h.peerDowns.Wait) {
|
if !returnsInTime(h.peerDowns.Wait) {
|
||||||
t.Fatalf("the PeersDown did not return within %s", stepTimeout)
|
t.Fatalf("the PeerDown did not return within %s", stepTimeout)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestPrefixHandlerTakesMessagesWhileAReturningPeerSendsItsTable gives a peer
|
|
||||||
// 50,000 prefixes and hands the prefix handler that peer's session going down.
|
|
||||||
// Once PeersDown has begun, the peer comes back and sends more announcements
|
|
||||||
// than the handler's queue holds, as a peer that sends its whole table again
|
|
||||||
// does. The handler must hold them back and go on taking messages. Another
|
|
||||||
// peer's announcement, handed next, must be applied before PeersDown ends.
|
|
||||||
func TestPrefixHandlerTakesMessagesWhileAReturningPeerSendsItsTable(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
const (
|
|
||||||
// downPeerPrefixes is five times the 10,000 prefixes PeersDown visits
|
|
||||||
// in one hold of the table's lock.
|
|
||||||
downPeerPrefixes = 50000
|
|
||||||
returningMessages = prefixHandlerQueueSize + 1
|
|
||||||
)
|
|
||||||
|
|
||||||
table := routetable.New(routetable.Params{})
|
|
||||||
announceManyPrefixes(table, downPeerPrefixes)
|
|
||||||
|
|
||||||
h := newPrefixHandler(newMockStore(), logger.New(), table,
|
|
||||||
longerThanAnyTest, maxTableChangesPerWrite)
|
|
||||||
defer stopInTime(t, h)
|
|
||||||
|
|
||||||
h.HandleMessage(sessionDown())
|
|
||||||
|
|
||||||
// Wait until PeersDown has begun: the down peer has left some of its
|
|
||||||
// prefixes, which are all IPv4.
|
|
||||||
for h.table.Counts().IPv4Routes == downPeerPrefixes {
|
|
||||||
runtime.Gosched()
|
|
||||||
}
|
|
||||||
|
|
||||||
returning := peerAnnouncement(testPeerIP, testV6Prefix)
|
|
||||||
|
|
||||||
if !returnsInTime(func() {
|
|
||||||
for range returningMessages {
|
|
||||||
h.HandleMessage(returning)
|
|
||||||
}
|
|
||||||
|
|
||||||
h.HandleMessage(peerAnnouncement(secondPeerIP, testV6Prefix))
|
|
||||||
}) {
|
|
||||||
t.Fatalf("the handler did not take the messages within %s", stepTimeout)
|
|
||||||
}
|
|
||||||
|
|
||||||
if h.table.Counts().IPv4Routes == 0 {
|
|
||||||
t.Error("PeersDown ended before the handler took the other peer's " +
|
|
||||||
"announcement")
|
|
||||||
}
|
|
||||||
|
|
||||||
checkTableOrigins(t, h.table, map[string]map[int][]string{
|
|
||||||
testV6Prefix: {testASNB: {secondPeerIP}},
|
|
||||||
})
|
|
||||||
|
|
||||||
if !returnsInTime(h.peerDowns.Wait) {
|
|
||||||
t.Fatalf("the PeersDown did not return within %s", stepTimeout)
|
|
||||||
}
|
|
||||||
|
|
||||||
checkTableOrigins(t, h.table, map[string]map[int][]string{
|
|
||||||
testV6Prefix: {testASNB: {testPeerIP, secondPeerIP}},
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes holds back many
|
// TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes holds back many
|
||||||
// messages of a peer whose session went down. Each names a prefix that does
|
// messages of a peer whose session went down. Each names a prefix that does
|
||||||
// not parse, so applying it logs an error. Each log write takes a
|
// not parse, so applying it logs an error. Each log write takes a
|
||||||
@@ -368,14 +237,14 @@ func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
|
|||||||
|
|
||||||
defer stopInTime(t, h)
|
defer stopInTime(t, h)
|
||||||
|
|
||||||
release := holdUpPeersDown(h)
|
h.peerDownMu.Lock()
|
||||||
h.HandleMessage(sessionDown())
|
h.HandleMessage(sessionDown())
|
||||||
|
|
||||||
for range heldMessages {
|
for range heldMessages {
|
||||||
h.HandleMessage(peerAnnouncement(testPeerIP, badPrefix))
|
h.HandleMessage(peerAnnouncement(testPeerIP, badPrefix))
|
||||||
}
|
}
|
||||||
|
|
||||||
close(release)
|
h.peerDownMu.Unlock()
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-logs.firstWrite:
|
case <-logs.firstWrite:
|
||||||
@@ -399,10 +268,10 @@ func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeersDown of a peer
|
// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeerDown of a peer
|
||||||
// whose session went down. It hands the handler maxHeldBackMessages messages
|
// whose session went down. It hands the handler maxHeldBackMessages messages
|
||||||
// of that peer. HandleMessage must then wait, even for another peer's
|
// of that peer. HandleMessage must then wait, even for another peer's
|
||||||
// message, until the PeersDown returns and a held-back message is applied.
|
// message, until the PeerDown returns and a held-back message is applied.
|
||||||
func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
|
func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
@@ -411,7 +280,7 @@ func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
|
|||||||
maxTableChangesPerWrite)
|
maxTableChangesPerWrite)
|
||||||
defer stopInTime(t, h)
|
defer stopInTime(t, h)
|
||||||
|
|
||||||
release := holdUpPeersDown(h)
|
h.peerDownMu.Lock()
|
||||||
h.HandleMessage(sessionDown())
|
h.HandleMessage(sessionDown())
|
||||||
|
|
||||||
held := peerAnnouncement(testPeerIP, testV6Prefix)
|
held := peerAnnouncement(testPeerIP, testV6Prefix)
|
||||||
@@ -433,12 +302,12 @@ func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
|
|||||||
case <-time.After(stillWaitingAfter):
|
case <-time.After(stillWaitingAfter):
|
||||||
}
|
}
|
||||||
|
|
||||||
close(release)
|
h.peerDownMu.Unlock()
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-returned:
|
case <-returned:
|
||||||
case <-time.After(stepTimeout):
|
case <-time.After(stepTimeout):
|
||||||
t.Fatal("HandleMessage did not return after the PeersDown returned")
|
t.Fatal("HandleMessage did not return after the PeerDown returned")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -555,7 +424,7 @@ func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// checkTableCases hands the messages of each case to a new prefix handler. It
|
// checkTableCases hands the messages of each case to a new prefix handler. It
|
||||||
// checks the handler's routing table once the handler's PeersDown calls have
|
// checks the handler's routing table once the handler's PeerDowns have
|
||||||
// returned, with no flush.
|
// returned, with no flush.
|
||||||
func checkTableCases(t *testing.T, tests []tableCase) {
|
func checkTableCases(t *testing.T, tests []tableCase) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|||||||
@@ -486,7 +486,6 @@ type HandlerStatsInfo struct {
|
|||||||
QueueHighWaterMark int `json:"queue_high_water_mark"`
|
QueueHighWaterMark int `json:"queue_high_water_mark"`
|
||||||
ProcessedCount uint64 `json:"processed_count"`
|
ProcessedCount uint64 `json:"processed_count"`
|
||||||
DroppedCount uint64 `json:"dropped_count"`
|
DroppedCount uint64 `json:"dropped_count"`
|
||||||
HeldBackCount int `json:"held_back_count"`
|
|
||||||
AvgProcessTimeMs float64 `json:"avg_process_time_ms"`
|
AvgProcessTimeMs float64 `json:"avg_process_time_ms"`
|
||||||
MinProcessTimeMs float64 `json:"min_process_time_ms"`
|
MinProcessTimeMs float64 `json:"min_process_time_ms"`
|
||||||
MaxProcessTimeMs float64 `json:"max_process_time_ms"`
|
MaxProcessTimeMs float64 `json:"max_process_time_ms"`
|
||||||
@@ -571,7 +570,6 @@ func handlerStatsInfo(handlerStats []streamer.HandlerStats) []HandlerStatsInfo {
|
|||||||
QueueHighWaterMark: hs.QueueHighWaterMark,
|
QueueHighWaterMark: hs.QueueHighWaterMark,
|
||||||
ProcessedCount: hs.ProcessedCount,
|
ProcessedCount: hs.ProcessedCount,
|
||||||
DroppedCount: hs.DroppedCount,
|
DroppedCount: hs.DroppedCount,
|
||||||
HeldBackCount: hs.HeldBackCount,
|
|
||||||
AvgProcessTimeMs: float64(hs.AvgProcessTime.Microseconds()) /
|
AvgProcessTimeMs: float64(hs.AvgProcessTime.Microseconds()) /
|
||||||
microsecondsPerMillisecond,
|
microsecondsPerMillisecond,
|
||||||
MinProcessTimeMs: float64(hs.MinProcessTime.Microseconds()) /
|
MinProcessTimeMs: float64(hs.MinProcessTime.Microseconds()) /
|
||||||
|
|||||||
@@ -295,43 +295,6 @@ func TestStatsAnswerCarriesChurnPerMinute(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// heldBackMessages is how many messages heldBackHandler holds back.
|
|
||||||
const heldBackMessages = 7
|
|
||||||
|
|
||||||
// heldBackHandler is a stream handler that wants no message and holds back
|
|
||||||
// heldBackMessages messages.
|
|
||||||
type heldBackHandler struct{}
|
|
||||||
|
|
||||||
func (heldBackHandler) WantsMessage(string) bool { return false }
|
|
||||||
func (heldBackHandler) HandleMessage(*ristypes.RISMessage) {}
|
|
||||||
func (heldBackHandler) QueueCapacity() int { return 1 }
|
|
||||||
func (heldBackHandler) HeldBackCount() int { return heldBackMessages }
|
|
||||||
|
|
||||||
// TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack registers a handler
|
|
||||||
// that holds back messages. Its figures in /api/v1/stats, which the status
|
|
||||||
// page shows, must give how many.
|
|
||||||
func TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
s := newServerWithTestRoutes(t)
|
|
||||||
s.streamer.RegisterHandler(heldBackHandler{})
|
|
||||||
|
|
||||||
var body struct {
|
|
||||||
Data StatsResponse `json:"data"`
|
|
||||||
}
|
|
||||||
|
|
||||||
err := json.Unmarshal(get(t, s, statsPath).Body.Bytes(), &body)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("decoding the answer: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
stats := body.Data.HandlerStats
|
|
||||||
if len(stats) != 1 || stats[0].HeldBackCount != heldBackMessages {
|
|
||||||
t.Errorf("handler figures %+v, want one handler holding back %d "+
|
|
||||||
"messages", stats, heldBackMessages)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// fixedIntervalASNFetcher is a WHOIS fetcher that does nothing and reports
|
// fixedIntervalASNFetcher is a WHOIS fetcher that does nothing and reports
|
||||||
// interval as its wait between fetches.
|
// interval as its wait between fetches.
|
||||||
type fixedIntervalASNFetcher struct {
|
type fixedIntervalASNFetcher struct {
|
||||||
|
|||||||
@@ -351,10 +351,6 @@ type HandlerStats struct {
|
|||||||
AvgProcessTime time.Duration
|
AvgProcessTime time.Duration
|
||||||
MinProcessTime time.Duration
|
MinProcessTime time.Duration
|
||||||
MaxProcessTime time.Duration
|
MaxProcessTime time.Duration
|
||||||
// HeldBackCount is how many messages the handler holds back to apply
|
|
||||||
// later: what its HeldBackCount method returns, or 0 for a handler
|
|
||||||
// without one.
|
|
||||||
HeldBackCount int
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetHandlerStats returns a snapshot of performance statistics for all registered
|
// GetHandlerStats returns a snapshot of performance statistics for all registered
|
||||||
@@ -396,11 +392,6 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
|
|||||||
|
|
||||||
info.metrics.mu.Unlock()
|
info.metrics.mu.Unlock()
|
||||||
|
|
||||||
holder, holdsBack := info.handler.(interface{ HeldBackCount() int })
|
|
||||||
if holdsBack {
|
|
||||||
hs.HeldBackCount = holder.HeldBackCount()
|
|
||||||
}
|
|
||||||
|
|
||||||
stats = append(stats, hs)
|
stats = append(stats, hs)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -541,10 +541,6 @@
|
|||||||
<span class="metric-label">Dropped</span>
|
<span class="metric-label">Dropped</span>
|
||||||
<span class="metric-value ${handler.dropped_count > 0 ? 'disconnected' : ''}">${formatNumber(handler.dropped_count)}</span>
|
<span class="metric-value ${handler.dropped_count > 0 ? 'disconnected' : ''}">${formatNumber(handler.dropped_count)}</span>
|
||||||
</div>
|
</div>
|
||||||
<div class="metric">
|
|
||||||
<span class="metric-label">Held Back</span>
|
|
||||||
<span class="metric-value">${formatNumber(handler.held_back_count)}</span>
|
|
||||||
</div>
|
|
||||||
<div class="metric">
|
<div class="metric">
|
||||||
<span class="metric-label">Avg Time</span>
|
<span class="metric-label">Avg Time</span>
|
||||||
<span class="metric-value">${formatProcessingTime(handler.avg_process_time_ms)}</span>
|
<span class="metric-value">${formatProcessingTime(handler.avg_process_time_ms)}</span>
|
||||||
|
|||||||
Reference in new issue
Block a user