diff --git a/README.md b/README.md index c97ee45..945dd13 100644 --- a/README.md +++ b/README.md @@ -445,8 +445,9 @@ the routing table, and everything else from the database. - A withdrawal removes the peer's route. So does an announcement whose AS path is empty or holds AS 0 anywhere. - A peer whose BGP session goes down leaves every prefix. The handler - holds back the peer's later updates until the peer has left every - prefix, then applies them in order. + holds back the peer's later updates. It applies them in order after + the peer has left every prefix and after the held-back updates of + every peer whose session went down before it. 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 @@ -651,10 +652,11 @@ What happens at each limit: at most about 0.1 GiB of the Go heap. - Held-back messages: the prefix handler holds back a peer's messages after the - peer's session goes down. It applies them once the peer has left every prefix. - It holds at most 20,000 such messages (`maxHeldBackMessages`), about 20 MiB. - At that number it takes nothing more off its queue until it has applied one of - them. + peer's session goes down. It applies them after the peer has left every prefix + and after the held-back messages of every peer whose session went down before + it. It holds at most 150,000 such messages (`maxHeldBackMessages`), about 150 + MiB. At that number it takes nothing more off its queue until it has applied + one of them. With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds with the goroutine count and Go memory figures. diff --git a/TODO.md b/TODO.md index d19cfd4..72b63d3 100644 --- a/TODO.md +++ b/TODO.md @@ -35,6 +35,14 @@ tested. # Completed Steps +- 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` diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index 33eabf1..f767034 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -273,7 +273,7 @@ func TestVacuumReturnsFreePages(t *testing.T) { writeTableChanges(t, db, table) // The peer going down removes every prefix, and so every row. - table.PeerDown(peerA) + table.PeersDown([]string{peerA}) writeTableChanges(t, db, table) var before int diff --git a/internal/database/routes_internal_test.go b/internal/database/routes_internal_test.go index 79bc133..711279f 100644 --- a/internal/database/routes_internal_test.go +++ b/internal/database/routes_internal_test.go @@ -107,7 +107,7 @@ func TestReadRouteTableGivesBackWhatWasWritten(t *testing.T) { // A peer moves to another origin, one goes down and a new one comes. announceTo(source, tablePrefixV4, asB, peerA, 60) - source.PeerDown(peerC) + source.PeersDown([]string{peerC}) announceTo(source, tablePrefixV6, asB, peerE, 70) writeTableChanges(t, db, source) @@ -133,7 +133,7 @@ func TestReadRouteTableGivesATableThatTakesLaterUpdates(t *testing.T) { // another one goes down, and a new peer comes. for _, table := range []*routetable.Table{source, read} { table.Withdraw(netip.MustParsePrefix(tablePrefixV4), peerA) - table.PeerDown(peerB) + table.PeersDown([]string{peerB}) announceTo(table, tablePrefixShortV4, asB, peerD, 50) } diff --git a/internal/routetable/reads_test.go b/internal/routetable/reads_test.go index bf4d470..6f61a88 100644 --- a/internal/routetable/reads_test.go +++ b/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, diff --git a/internal/routetable/routetable.go b/internal/routetable/routetable.go index b083b6c..339df90 100644 --- a/internal/routetable/routetable.go +++ b/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 diff --git a/internal/routetable/routetable_internal_test.go b/internal/routetable/routetable_internal_test.go index 64dde1a..2f3b286 100644 --- a/internal/routetable/routetable_internal_test.go +++ b/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: } } diff --git a/internal/routetable/routetable_test.go b/internal/routetable/routetable_test.go index c2aea91..f2ff722 100644 --- a/internal/routetable/routetable_test.go +++ b/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()) }, diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 666edf3..c7068c3 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -46,10 +46,19 @@ const ( maxTableChangesPerWrite = 20000 // maxHeldBackMessages is the most messages the handler holds back from - // peers whose PeerDown has not returned. A message takes about 1 KiB, so - // that many take about 20 MiB (README "Memory"). At this number - // HandleMessage waits until one of them is applied. - maxHeldBackMessages = 20000 + // peers whose session went down. A peer's messages are held back from + // the moment its session goes down until peerDown has applied them. If + // peerDown is running at that moment, the peer waits for the rest of + // peerDown's current PeersDown call. The peer also waits while peerDown + // 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 @@ -58,14 +67,16 @@ const ( // does an announcement whose AS path is empty or holds AS 0 anywhere. A peer // whose BGP session goes down leaves 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 PeerDown on -// a separate goroutine, one peer at a time, and goes on taking messages -// meanwhile. PeerDown lets the table's lock go about half of the time, so the -// handler applies other peers' messages while PeerDown runs. The handler holds -// back the down peer's later messages until PeerDown returns. Then the handler -// applies them in order, one at a time, and takes new messages in between. It -// holds back at most maxHeldBackMessages messages. +// The table's PeersDown applies sessions going down. It visits every prefix, +// which takes seconds on a full table. So the handler runs PeersDown on a +// separate goroutine and goes on taking messages meanwhile. One PeersDown runs +// at a time. The next one takes every peer whose session went down since the +// previous one began. PeersDown lets the table's lock go about half of the +// time, so the handler applies other peers' messages while PeersDown runs. The +// handler holds back a down peer's later messages until peerDown applies them, +// as the doc of maxHeldBackMessages describes. It applies them in order, one at +// 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 // prefix the updates name with the time it was first seen. @@ -103,20 +114,25 @@ type PrefixHandler struct { stopCh chan context.Context done chan struct{} // closed when the flush goroutine returns - // Peers whose session went down. peerDownMu lets one PeerDown run at a - // time, so the handler's own updates wait for the table's lock behind - // at most one. heldBack, under mu, holds each such peer's later - // 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. + // Peers whose session went down. heldBack, under mu, holds each such + // peer's later messages, in order, until the peer has left every prefix + // 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 - // be applied; its L is &mu. peerDowns counts the peerDown goroutines - // that have not returned. - peerDownMu sync.Mutex - heldBack map[string][]*ristypes.RISMessage - heldCount int - heldTaken sync.Cond - peerDowns sync.WaitGroup + // be applied; its L is &mu. + heldBack map[string][]*ristypes.RISMessage + heldCount int + heldTaken sync.Cond + // downPeers, under mu, lists the peers whose session went down and that + // the next PeersDown takes. peerDownRunning, under mu, says whether + // peerDown is running. peerDowns counts the peerDown goroutines that have + // 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 @@ -181,6 +197,7 @@ func newPrefixHandler( stopCh: make(chan context.Context, 1), done: make(chan struct{}), heldBack: make(map[string][]*ristypes.RISMessage), + tablePeersDown: table.PeersDown, } h.batchTaken.L = &h.mu h.heldTaken.L = &h.mu @@ -212,7 +229,7 @@ func (h *PrefixHandler) QueueCapacity() int { } // HandleMessage processes a message to track prefix information. A message -// from a peer whose PeerDown has not returned is held back until it has. +// from a peer whose session went down is held back until peerDown applies it. func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { h.mu.Lock() defer h.mu.Unlock() @@ -235,7 +252,16 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { h.applyMessage(msg) } -// Stop waits for every PeerDown the handler started and for the messages held +// HeldBackCount returns how many messages the handler holds back from peers +// 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 // 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 @@ -290,8 +316,8 @@ func (h *PrefixHandler) Stop(ctx context.Context) { } // applyMessage applies msg. The caller holds mu. A peer's session going down -// starts the peer's PeerDown. An update is applied to the routing table, and -// its prefixes join the batch. +// has the peer leave every prefix. An update is applied to the routing table, +// and its prefixes join the batch. func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) { if msg.Type == ristypes.MessageTypeState { if msg.State == ristypes.StateDown { @@ -370,34 +396,64 @@ func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) { } } -// startPeerDown starts peerDown for peer on a new goroutine. The caller holds -// mu. The peer's messages are held back until peerDown has applied them. -// Messages held back for an earlier PeerDown of the peer, and not yet applied, -// stay held back. +// startPeerDown adds peer to downPeers. It starts peerDown on a new goroutine +// unless peerDown is running. The caller holds mu. The peer's messages are +// held back until peerDown has applied them. Messages held back for an earlier +// session of the peer going down, and not yet applied, stay held back. func (h *PrefixHandler) startPeerDown(peer string) { _, goingDown := h.heldBack[peer] if !goingDown { h.heldBack[peer] = []*ristypes.RISMessage{} } + h.downPeers = append(h.downPeers, peer) + + if h.peerDownRunning { + return + } + + h.peerDownRunning = true h.peerDowns.Add(1) - go h.peerDown(peer) + go h.peerDown() } -// peerDown runs the table's PeerDown for peer, once no other PeerDown is -// running. Then peerDown applies the messages held back from the 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 -// starts a new peerDown, which applies the rest. -func (h *PrefixHandler) peerDown(peer string) { +// peerDown takes every peer in downPeers and runs the table's PeersDown for +// them. Then it applies the messages held back from each of those peers. It +// repeats this until downPeers is empty. +func (h *PrefixHandler) peerDown() { defer h.peerDowns.Done() - h.peerDownMu.Lock() - h.table.PeerDown(peer) - h.peerDownMu.Unlock() + for { + h.mu.Lock() + 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 { h.mu.Lock() diff --git a/internal/routewatch/prefixhandler_internal_test.go b/internal/routewatch/prefixhandler_internal_test.go index 10c190f..d7ce0a7 100644 --- a/internal/routewatch/prefixhandler_internal_test.go +++ b/internal/routewatch/prefixhandler_internal_test.go @@ -238,7 +238,7 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) { } // TestPrefixHandlerStopLogsTheMessagesHeldBack stops a prefix handler with a -// deadline that has passed, while a peer's PeerDown is held up and one message +// deadline that has passed, while a peer's PeersDown is held up and one message // of the peer is held back. Stop must log that message as not applied. func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) { t.Parallel() @@ -249,8 +249,8 @@ func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) { routetable.New(routetable.Params{}), longerThanAnyTest, maxTableChangesPerWrite) - h.peerDownMu.Lock() - defer h.peerDownMu.Unlock() + release := holdUpPeersDown(h) + defer close(release) h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix)) h.HandleMessage(sessionDown()) @@ -268,8 +268,8 @@ func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) { } // TestPrefixHandlerStopWaitsForTheMessagesHeldBack stops a prefix handler -// while a peer's PeerDown is held up and an announcement of the peer is held -// back. Stop must wait. Once the PeerDown is let go and Stop returns, the +// while a peer's PeersDown 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 // announcement must be in the database. func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) { t.Parallel() @@ -278,7 +278,7 @@ func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) { h := newPrefixHandler(db, logger.New(), routetable.New(routetable.Params{}), longerThanAnyTest, maxTableChangesPerWrite) - h.peerDownMu.Lock() + release := holdUpPeersDown(h) h.HandleMessage(sessionDown()) h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix)) @@ -291,16 +291,16 @@ func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) { select { case <-stopped: - t.Error("Stop returned while a PeerDown was held up") + t.Error("Stop returned while a PeersDown was held up") case <-time.After(stillWaitingAfter): } - h.peerDownMu.Unlock() + close(release) select { case <-stopped: case <-time.After(stepTimeout): - t.Fatalf("Stop did not return within %s of the PeerDown's release", + t.Fatalf("Stop did not return within %s of the PeersDown's release", stepTimeout) } @@ -324,16 +324,32 @@ const longerThanAnyTest = time.Hour // applies the updates to an empty routing table. It is built without // NewPrefixHandler's flush goroutine, so the test flushes itself. func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler { + table := routetable.New(routetable.Params{}) + return &PrefixHandler{ db: store, logger: logger.New(), - table: routetable.New(routetable.Params{}), + table: table, batchMaxSize: prefixBatchMaxSize, tableChangesPerWrite: maxTableChangesPerWrite, 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 // when the test ends. func newTestDatabase(t *testing.T) *database.Database { diff --git a/internal/routewatch/prefixhandler_table_internal_test.go b/internal/routewatch/prefixhandler_table_internal_test.go index fa94967..65b3003 100644 --- a/internal/routewatch/prefixhandler_table_internal_test.go +++ b/internal/routewatch/prefixhandler_table_internal_test.go @@ -103,23 +103,24 @@ func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) { }) } -// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeerDown of a -// peer whose session went down, as another peer's long PeerDown would. The +// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeersDown of a +// peer whose session went down, as another peer's long PeersDown would. The // 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: // - an announcement; // - its session going down again; // - a second announcement. // -// Once the PeerDowns have returned, the down peer must carry only the prefix -// of the second announcement. +// HeldBackCount must count them. Once the PeersDown calls have returned, the +// down peer must carry only the prefix of the second announcement. func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) { t.Parallel() - h := prefixHandlerWithoutTimer(newMockStore()) - h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix)) + const heldMessages = 3 - h.peerDownMu.Lock() + h := prefixHandlerWithoutTimer(newMockStore()) + release := holdUpPeersDown(h) + h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix)) handled := returnsInTime(func() { for _, msg := range []*ristypes.RISMessage{ @@ -144,10 +145,15 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) { setV4Prefix: {}, }) - h.peerDownMu.Unlock() + if h.HeldBackCount() != heldMessages { + t.Errorf("HeldBackCount() = %d, want %d", h.HeldBackCount(), + heldMessages) + } + + close(release) if !returnsInTime(h.peerDowns.Wait) { - t.Fatalf("the PeerDowns did not return within %s", stepTimeout) + t.Fatalf("the PeersDown calls did not return within %s", stepTimeout) } checkTableOrigins(t, h.table, map[string]map[int][]string{ @@ -158,21 +164,85 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) { }) } -// TestPrefixHandlerAppliesMessagesWhilePeerDownRuns gives a peer 50,000 +// TestPrefixHandlerTakesPeersDownTogether has four peers announce testV4Prefix. +// 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 -// table's PeerDown then visits the prefixes in five batches. Once PeerDown has -// begun, the handler is handed 1,000 announcements of another peer. The -// handler must apply them all before PeerDown ends, while the down peer still +// table's PeersDown then visits the prefixes in five batches. Once PeersDown +// has 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 // carries some of its prefixes. -func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) { +func TestPrefixHandlerAppliesMessagesWhilePeersDownRuns(t *testing.T) { t.Parallel() const ( - // downPeerPrefixes is five times the 10,000 prefixes PeerDown visits + // downPeerPrefixes is five times the 10,000 prefixes PeersDown visits // in one hold of the table's lock. downPeerPrefixes = 50000 // otherPeerMessages is far more messages than the handler applies - // while PeerDown runs, if PeerDown takes the lock back at once after + // while PeersDown runs, if PeersDown takes the lock back at once after // each batch. The handler then applies only a few for each batch. otherPeerMessages = 1000 // firstOtherAddr is the address of the other peer's first prefix. @@ -184,7 +254,7 @@ func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) { h.HandleMessage(sessionDown()) - // Wait until PeerDown has begun: the down peer has left some of its + // Wait until PeersDown has begun: the down peer has left some of its // prefixes. The down peer's prefixes are IPv4 and the other peer's are // IPv6, so the counts tell them apart. for h.table.Counts().IPv4Routes == downPeerPrefixes { @@ -207,15 +277,76 @@ func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) { } if counts.IPv4Routes == 0 { - t.Errorf("PeerDown ended before the handler applied the other peer's "+ + t.Errorf("PeersDown ended before the handler applied the other peer's "+ "%d announcements", otherPeerMessages) } if !returnsInTime(h.peerDowns.Wait) { - t.Fatalf("the PeerDown did not return within %s", stepTimeout) + t.Fatalf("the PeersDown 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 // 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 @@ -237,14 +368,14 @@ func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) { defer stopInTime(t, h) - h.peerDownMu.Lock() + release := holdUpPeersDown(h) h.HandleMessage(sessionDown()) for range heldMessages { h.HandleMessage(peerAnnouncement(testPeerIP, badPrefix)) } - h.peerDownMu.Unlock() + close(release) select { case <-logs.firstWrite: @@ -268,10 +399,10 @@ func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) { } } -// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeerDown of a peer +// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeersDown of a peer // whose session went down. It hands the handler maxHeldBackMessages messages // of that peer. HandleMessage must then wait, even for another peer's -// message, until the PeerDown returns and a held-back message is applied. +// message, until the PeersDown returns and a held-back message is applied. func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) { t.Parallel() @@ -280,7 +411,7 @@ func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) { maxTableChangesPerWrite) defer stopInTime(t, h) - h.peerDownMu.Lock() + release := holdUpPeersDown(h) h.HandleMessage(sessionDown()) held := peerAnnouncement(testPeerIP, testV6Prefix) @@ -302,12 +433,12 @@ func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) { case <-time.After(stillWaitingAfter): } - h.peerDownMu.Unlock() + close(release) select { case <-returned: case <-time.After(stepTimeout): - t.Fatal("HandleMessage did not return after the PeerDown returned") + t.Fatal("HandleMessage did not return after the PeersDown returned") } } @@ -424,7 +555,7 @@ func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) { } // checkTableCases hands the messages of each case to a new prefix handler. It -// checks the handler's routing table once the handler's PeerDowns have +// checks the handler's routing table once the handler's PeersDown calls have // returned, with no flush. func checkTableCases(t *testing.T, tests []tableCase) { t.Helper() diff --git a/internal/server/handlers.go b/internal/server/handlers.go index cf9b9bc..8ae2d90 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -486,6 +486,7 @@ type HandlerStatsInfo struct { QueueHighWaterMark int `json:"queue_high_water_mark"` ProcessedCount uint64 `json:"processed_count"` DroppedCount uint64 `json:"dropped_count"` + HeldBackCount int `json:"held_back_count"` AvgProcessTimeMs float64 `json:"avg_process_time_ms"` MinProcessTimeMs float64 `json:"min_process_time_ms"` MaxProcessTimeMs float64 `json:"max_process_time_ms"` @@ -570,6 +571,7 @@ func handlerStatsInfo(handlerStats []streamer.HandlerStats) []HandlerStatsInfo { QueueHighWaterMark: hs.QueueHighWaterMark, ProcessedCount: hs.ProcessedCount, DroppedCount: hs.DroppedCount, + HeldBackCount: hs.HeldBackCount, AvgProcessTimeMs: float64(hs.AvgProcessTime.Microseconds()) / microsecondsPerMillisecond, MinProcessTimeMs: float64(hs.MinProcessTime.Microseconds()) / diff --git a/internal/server/handlers_internal_test.go b/internal/server/handlers_internal_test.go index 3ae47f1..d2f4654 100644 --- a/internal/server/handlers_internal_test.go +++ b/internal/server/handlers_internal_test.go @@ -295,6 +295,43 @@ 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 // interval as its wait between fetches. type fixedIntervalASNFetcher struct { diff --git a/internal/streamer/streamer.go b/internal/streamer/streamer.go index 5beaf95..b90b8cc 100644 --- a/internal/streamer/streamer.go +++ b/internal/streamer/streamer.go @@ -351,6 +351,10 @@ type HandlerStats struct { AvgProcessTime time.Duration MinProcessTime 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 @@ -392,6 +396,11 @@ func (s *Streamer) GetHandlerStats() []HandlerStats { info.metrics.mu.Unlock() + holder, holdsBack := info.handler.(interface{ HeldBackCount() int }) + if holdsBack { + hs.HeldBackCount = holder.HeldBackCount() + } + stats = append(stats, hs) } diff --git a/internal/templates/status.html b/internal/templates/status.html index 3463f4a..f3959d7 100644 --- a/internal/templates/status.html +++ b/internal/templates/status.html @@ -541,6 +541,10 @@ Dropped ${formatNumber(handler.dropped_count)} +