Start no prefix flush after the stop deadline (closes #188) #190
@@ -33,6 +33,9 @@ tested.
|
|||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
|
- 2026-10-27: once the stop deadline has passed, the prefix handler starts no
|
||||||
|
flush. A batch that filled while a write ran past the deadline could still be
|
||||||
|
written after it (closes #188)
|
||||||
- 2026-10-27: the prefix handler applies each update to the routing table in
|
- 2026-10-27: the prefix handler applies each update to the routing table in
|
||||||
memory, and writes the table's changes to the database every second. A write
|
memory, and writes the table's changes to the database every second. A write
|
||||||
takes at most 20,000 changed pairs, and the rest follow in the next writes. A
|
takes at most 20,000 changed pairs, and the rest follow in the next writes. A
|
||||||
|
|||||||
@@ -317,7 +317,9 @@ func stopWhileAWriteIsBlocked(
|
|||||||
|
|
||||||
// TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline also checks the
|
// TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline also checks the
|
||||||
// routing table's changes: the flush whose delete is held wrote those of the
|
// routing table's changes: the flush whose delete is held wrote those of the
|
||||||
// fill, so Stop must log the one of the extra prefix as not written.
|
// fill, so Stop must log the one of the extra prefix as not written. The flush
|
||||||
|
// timer must not fire. A flush it started before the fill would be the one
|
||||||
|
// held, and it would not have written the fill's change.
|
||||||
func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
|
func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
@@ -326,8 +328,8 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
|
|||||||
var logs bytes.Buffer
|
var logs bytes.Buffer
|
||||||
|
|
||||||
store := newBlockingStore()
|
store := newBlockingStore()
|
||||||
h := NewPrefixHandler(store, newErrorLogger(&logs),
|
h := newPrefixHandler(store, newErrorLogger(&logs),
|
||||||
routetable.New(routetable.Params{}))
|
routetable.New(routetable.Params{}), longerThanAnyTest)
|
||||||
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
|
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
|
||||||
// The peer's session goes down first, so the write held is the delete of
|
// The peer's session goes down first, so the write held is the delete of
|
||||||
// its routes, and Stop must log that peer down too.
|
// its routes, and Stop must log that peer down too.
|
||||||
@@ -351,6 +353,31 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestPrefixHandlerSkipsAFullBatchAfterTheStopDeadline hands the prefix handler
|
||||||
|
// a full batch while a write is held, so the batch's wake-up of the flush
|
||||||
|
// goroutine waits. The stop deadline passes meanwhile. When the write ends, the
|
||||||
|
// flush goroutine must stop instead of writing the batch.
|
||||||
|
func TestPrefixHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
const extraPrefix = "198.51.100.0/24"
|
||||||
|
|
||||||
|
store := newBlockingStore()
|
||||||
|
h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}))
|
||||||
|
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
|
||||||
|
full := slices.Repeat([]string{extraPrefix}, prefixBatchSize)
|
||||||
|
stopWhileAWriteIsBlocked(t, store, h, h.done,
|
||||||
|
[]*ristypes.RISMessage{prefixAnnouncement(fill)},
|
||||||
|
prefixAnnouncement(full))
|
||||||
|
|
||||||
|
store.mu.Lock()
|
||||||
|
defer store.mu.Unlock()
|
||||||
|
|
||||||
|
if store.prefixes[extraPrefix] {
|
||||||
|
t.Errorf("%s was written after the stop deadline", extraPrefix)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// tableWriteBlockingStore is a blockingStore whose writes of the routing
|
// tableWriteBlockingStore is a blockingStore whose writes of the routing
|
||||||
// table's changes also wait until the test calls releaseWrites.
|
// table's changes also wait until the test calls releaseWrites.
|
||||||
type tableWriteBlockingStore struct {
|
type tableWriteBlockingStore struct {
|
||||||
|
|||||||
@@ -130,6 +130,16 @@ func (u prefixUpdate) announcesRoute() bool {
|
|||||||
// the updates to table, the routing table read from db at startup.
|
// the updates to table, the routing table read from db at startup.
|
||||||
func NewPrefixHandler(
|
func NewPrefixHandler(
|
||||||
db database.Store, logger *logger.Logger, table *routetable.Table,
|
db database.Store, logger *logger.Logger, table *routetable.Table,
|
||||||
|
) *PrefixHandler {
|
||||||
|
return newPrefixHandler(db, logger, table, prefixBatchTimeout)
|
||||||
|
}
|
||||||
|
|
||||||
|
// newPrefixHandler is NewPrefixHandler with the flush timer's interval,
|
||||||
|
// batchTimeout, in place of prefixBatchTimeout. A test that the timer must not
|
||||||
|
// disturb gives an interval longer than the test runs.
|
||||||
|
func newPrefixHandler(
|
||||||
|
db database.Store, logger *logger.Logger, table *routetable.Table,
|
||||||
|
batchTimeout time.Duration,
|
||||||
) *PrefixHandler {
|
) *PrefixHandler {
|
||||||
h := &PrefixHandler{
|
h := &PrefixHandler{
|
||||||
db: db,
|
db: db,
|
||||||
@@ -144,7 +154,7 @@ func NewPrefixHandler(
|
|||||||
h.batchTaken.L = &h.mu
|
h.batchTaken.L = &h.mu
|
||||||
|
|
||||||
// Start the flush timer goroutine
|
// Start the flush timer goroutine
|
||||||
go h.flushLoop()
|
go h.flushLoop(batchTimeout)
|
||||||
|
|
||||||
return h
|
return h
|
||||||
}
|
}
|
||||||
@@ -315,17 +325,32 @@ func (h *PrefixHandler) peerSessionDown(peer string) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// flushLoop runs in a goroutine, the only one that writes the batch: it
|
// flushLoop runs in a goroutine, the only one that writes the batch: it
|
||||||
// flushes when the batch is full, periodically, and once more on Stop.
|
// flushes when the batch is full, every batchTimeout, and once more on Stop.
|
||||||
func (h *PrefixHandler) flushLoop() {
|
func (h *PrefixHandler) flushLoop(batchTimeout time.Duration) {
|
||||||
defer close(h.done)
|
defer close(h.done)
|
||||||
|
|
||||||
ticker := time.NewTicker(prefixBatchTimeout)
|
ticker := time.NewTicker(batchTimeout)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
for {
|
for {
|
||||||
|
// Stop goes first. A write can run past the stop deadline while a full
|
||||||
|
// batch's wake-up or a tick waits beside Stop's context. The select
|
||||||
|
// below picks one of them at random, but past the deadline no flush
|
||||||
|
// may start.
|
||||||
|
select {
|
||||||
|
case ctx := <-h.stopCh:
|
||||||
|
// Final flush, unless the stop deadline has passed
|
||||||
|
if ctx.Err() == nil {
|
||||||
|
h.flushBatch()
|
||||||
|
}
|
||||||
|
|
||||||
|
return
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
if time.Since(h.lastFlush) >= prefixBatchTimeout {
|
if time.Since(h.lastFlush) >= batchTimeout {
|
||||||
h.flushBatch()
|
h.flushBatch()
|
||||||
}
|
}
|
||||||
case <-h.batchFullCh:
|
case <-h.batchFullCh:
|
||||||
|
|||||||
@@ -980,6 +980,11 @@ func TestPrefixHandlerStopLogsNothingAfterADeleteThatSucceeds(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// longerThanAnyTest is a flush timer interval that no test lasts. A test whose
|
||||||
|
// prefix handler must flush only when its batch is full and on Stop gives it to
|
||||||
|
// newPrefixHandler.
|
||||||
|
const longerThanAnyTest = time.Hour
|
||||||
|
|
||||||
// prefixHandlerWithoutTimer returns a prefix handler that writes to store and
|
// prefixHandlerWithoutTimer returns a prefix handler that writes to store and
|
||||||
// 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.
|
||||||
|
|||||||
@@ -341,15 +341,14 @@ func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten stops a prefix handler,
|
// TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten stops a prefix handler,
|
||||||
// with no deadline, whose routing table holds more changed pairs than two
|
// with no deadline, whose routing table holds more changed pairs than one write
|
||||||
// writes take. Its last flush writes one write's worth, so Stop must log the
|
// takes. Its last flush writes one write's worth, so Stop must log the pairs
|
||||||
// pairs that no write took.
|
// that no write took. The flush timer must not fire, or its writes could leave
|
||||||
|
// no pairs for Stop to log.
|
||||||
func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
|
func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
// More than two writes take, so that pairs are left even when the flush
|
const pairs = maxTableChangesPerWrite + 1
|
||||||
// timer writes once before Stop.
|
|
||||||
const pairs = 2*maxTableChangesPerWrite + 1
|
|
||||||
|
|
||||||
var logs bytes.Buffer
|
var logs bytes.Buffer
|
||||||
|
|
||||||
@@ -357,7 +356,7 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
|
|||||||
table := routetable.New(routetable.Params{})
|
table := routetable.New(routetable.Params{})
|
||||||
announceManyPrefixes(table, pairs)
|
announceManyPrefixes(table, pairs)
|
||||||
|
|
||||||
h := NewPrefixHandler(store, newErrorLogger(&logs), table)
|
h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest)
|
||||||
h.Stop(t.Context())
|
h.Stop(t.Context())
|
||||||
|
|
||||||
written := 0
|
written := 0
|
||||||
|
|||||||
Reference in New Issue
Block a user