The prefix, AS and peer handlers skipped a timed write unless one interval had passed since the last write ended. The next tick came a little less than one interval after that, so every other tick was skipped. Each tick now writes. A full batch's write restarts the timer, so the next timed write comes one interval after that write began. Comparing clock readings would not do: two ticks one interval apart can read a hair under it. flushBatch no longer reports whether it wrote a batch, since nothing reads that. A new test runs the loop on a fake clock and fails on the old code. Model: opus-5-5
This commit is contained in:
@@ -33,6 +33,10 @@ tested.
|
|||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
|
- 2026-10-27: the prefix, AS and peer handlers make a timed write on every tick
|
||||||
|
of their timers, every 1, 2 and 2 seconds. They used to skip every other tick.
|
||||||
|
A full batch's write restarts the timer, so the next timed write comes one
|
||||||
|
interval after that write began (closes #192)
|
||||||
- 2026-10-27: once the stop deadline has passed, the AS, peer and peering
|
- 2026-10-27: once the stop deadline has passed, the AS, peer and peering
|
||||||
handlers start no write either. A full batch or a tick of the timer that
|
handlers start no write either. A full batch or a tick of the timer that
|
||||||
waited while a write ran past the deadline could still start one. The flush
|
waited while a write ran past the deadline could still start one. The flush
|
||||||
|
|||||||
@@ -202,11 +202,11 @@ func (h *ASHandler) takeBatch() []asnOp {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// flushBatch takes the pending batch and writes it to the database, without
|
// flushBatch takes the pending batch and writes it to the database, without
|
||||||
// holding the mutex during the write. It reports whether there was a batch.
|
// holding the mutex during the write.
|
||||||
func (h *ASHandler) flushBatch() bool {
|
func (h *ASHandler) flushBatch() {
|
||||||
batch := h.takeBatch()
|
batch := h.takeBatch()
|
||||||
if len(batch) == 0 {
|
if len(batch) == 0 {
|
||||||
return false
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// The earliest and the latest time the batch saw each AS. The batch holds
|
// The earliest and the latest time the batch saw each AS. The batch holds
|
||||||
@@ -229,6 +229,4 @@ func (h *ASHandler) flushBatch() bool {
|
|||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
h.writing = 0
|
h.writing = 0
|
||||||
h.mu.Unlock()
|
h.mu.Unlock()
|
||||||
|
|
||||||
return true
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -6,27 +6,26 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// flushLoop is the flush goroutine of the prefix, AS and peer handlers. It is
|
// flushLoop is the flush goroutine of the prefix, AS and peer handlers. It is
|
||||||
// the only goroutine that writes a handler's batch. flushBatch takes the batch,
|
// the only goroutine that writes a handler's batch. flushBatch takes the batch
|
||||||
// writes it, and reports whether there was one.
|
// and writes it.
|
||||||
//
|
//
|
||||||
// flushLoop calls flushBatch when batchFullCh wakes it. It also calls it every
|
// flushLoop calls flushBatch on each tick of a timer, every batchTimeout. It
|
||||||
// batchTimeout, unless it wrote a batch less than batchTimeout ago. When stopCh
|
// also calls it at once when batchFullCh wakes it. A full batch's write
|
||||||
// hands it Stop's context, it calls flushBatch once more, unless the stop
|
// restarts the timer, so the next tick comes batchTimeout after that write
|
||||||
// deadline has passed. Then it closes done and returns.
|
// began. When stopCh hands it Stop's context, it calls flushBatch once more,
|
||||||
|
// unless the stop deadline has passed. Then it closes done and returns.
|
||||||
func flushLoop(
|
func flushLoop(
|
||||||
batchTimeout time.Duration,
|
batchTimeout time.Duration,
|
||||||
batchFullCh <-chan struct{},
|
batchFullCh <-chan struct{},
|
||||||
stopCh <-chan context.Context,
|
stopCh <-chan context.Context,
|
||||||
done chan<- struct{},
|
done chan<- struct{},
|
||||||
flushBatch func() bool,
|
flushBatch func(),
|
||||||
) {
|
) {
|
||||||
defer close(done)
|
defer close(done)
|
||||||
|
|
||||||
ticker := time.NewTicker(batchTimeout)
|
ticker := time.NewTicker(batchTimeout)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
lastFlush := time.Now()
|
|
||||||
|
|
||||||
for {
|
for {
|
||||||
// Stop goes first. A write can run past the stop deadline while a full
|
// 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
|
// batch's wake-up or a tick waits beside Stop's context. The select
|
||||||
@@ -45,13 +44,12 @@ func flushLoop(
|
|||||||
|
|
||||||
select {
|
select {
|
||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
if time.Since(lastFlush) >= batchTimeout && flushBatch() {
|
flushBatch()
|
||||||
lastFlush = time.Now()
|
|
||||||
}
|
|
||||||
case <-batchFullCh:
|
case <-batchFullCh:
|
||||||
if flushBatch() {
|
// Reset also drops a tick that waits in ticker.C, so no timed
|
||||||
lastFlush = time.Now()
|
// write follows this one at once.
|
||||||
}
|
ticker.Reset(batchTimeout)
|
||||||
|
flushBatch()
|
||||||
case ctx := <-stopCh:
|
case ctx := <-stopCh:
|
||||||
// Final flush, unless the stop deadline has passed
|
// Final flush, unless the stop deadline has passed
|
||||||
if ctx.Err() == nil {
|
if ctx.Err() == nil {
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import (
|
|||||||
"slices"
|
"slices"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
|
"testing/synctest"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.eeqj.de/sneak/routewatch/internal/database"
|
"git.eeqj.de/sneak/routewatch/internal/database"
|
||||||
@@ -27,6 +28,58 @@ const (
|
|||||||
stillWaitingAfter = 100 * time.Millisecond
|
stillWaitingAfter = 100 * time.Millisecond
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// TestFlushLoopWritesOnEveryTick runs flushLoop in a synctest bubble, whose
|
||||||
|
// clock is fake: time.Sleep there moves the clock on and returns at once. Each
|
||||||
|
// write takes a tenth of the interval. Two timed writes must start one interval
|
||||||
|
// apart. A full batch's write, which starts as the second timed write ends,
|
||||||
|
// must be followed by a timed write one interval after it began.
|
||||||
|
func TestFlushLoopWritesOnEveryTick(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
synctest.Test(t, func(t *testing.T) {
|
||||||
|
const (
|
||||||
|
interval = time.Second
|
||||||
|
writeTime = 100 * time.Millisecond
|
||||||
|
)
|
||||||
|
|
||||||
|
writeStarts := make(chan time.Time)
|
||||||
|
batchFullCh := make(chan struct{}, 1)
|
||||||
|
stopCh := make(chan context.Context, 1)
|
||||||
|
done := make(chan struct{})
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
flushLoop(interval, batchFullCh, stopCh, done, func() {
|
||||||
|
writeStarts <- time.Now()
|
||||||
|
|
||||||
|
time.Sleep(writeTime)
|
||||||
|
})
|
||||||
|
}()
|
||||||
|
|
||||||
|
first := <-writeStarts
|
||||||
|
second := <-writeStarts
|
||||||
|
|
||||||
|
if gap := second.Sub(first); gap != interval {
|
||||||
|
t.Errorf("the second timed write started %s after the first, "+
|
||||||
|
"want %s", gap, interval)
|
||||||
|
}
|
||||||
|
|
||||||
|
batchFullCh <- struct{}{}
|
||||||
|
|
||||||
|
full := <-writeStarts
|
||||||
|
timed := <-writeStarts
|
||||||
|
|
||||||
|
if gap := timed.Sub(full); gap != interval {
|
||||||
|
t.Errorf("the timed write after a full batch's started %s after "+
|
||||||
|
"it, want %s", gap, interval)
|
||||||
|
}
|
||||||
|
|
||||||
|
stopCh <- t.Context()
|
||||||
|
|
||||||
|
<-writeStarts // Stop's last write
|
||||||
|
<-done
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
// blockingStore is a mockStore whose batch writes for the AS, peer and prefix
|
// blockingStore is a mockStore whose batch writes for the AS, peer and prefix
|
||||||
// handlers, delete of a peer's routes and peering write wait until the test
|
// handlers, delete of a peer's routes and peering write wait until the test
|
||||||
// calls releaseWrites. writeStarted is closed when the first of them starts. It
|
// calls releaseWrites. writeStarted is closed when the first of them starts. It
|
||||||
|
|||||||
@@ -200,11 +200,11 @@ func (h *PeerHandler) takeBatch() []peerUpdate {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// flushBatch takes the pending batch and writes it to the database, without
|
// flushBatch takes the pending batch and writes it to the database, without
|
||||||
// holding the mutex during the write. It reports whether there was a batch.
|
// holding the mutex during the write.
|
||||||
func (h *PeerHandler) flushBatch() bool {
|
func (h *PeerHandler) flushBatch() {
|
||||||
batch := h.takeBatch()
|
batch := h.takeBatch()
|
||||||
if len(batch) == 0 {
|
if len(batch) == 0 {
|
||||||
return false
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// The earliest and the latest time the batch saw each peer, with the peer
|
// The earliest and the latest time the batch saw each peer, with the peer
|
||||||
@@ -234,6 +234,4 @@ func (h *PeerHandler) flushBatch() bool {
|
|||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
h.writing = 0
|
h.writing = 0
|
||||||
h.mu.Unlock()
|
h.mu.Unlock()
|
||||||
|
|
||||||
return true
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -352,13 +352,12 @@ func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) {
|
|||||||
|
|
||||||
// flushBatch writes the routing table's changes, then takes the pending batch
|
// flushBatch writes the routing table's changes, then takes the pending batch
|
||||||
// and writes it to the database, without holding the mutex during the writes.
|
// and writes it to the database, without holding the mutex during the writes.
|
||||||
// It reports whether there was a batch, or a peer down to delete the routes of.
|
func (h *PrefixHandler) flushBatch() {
|
||||||
func (h *PrefixHandler) flushBatch() bool {
|
|
||||||
h.writeTableChanges()
|
h.writeTableChanges()
|
||||||
|
|
||||||
batch, peersDown := h.takeBatch()
|
batch, peersDown := h.takeBatch()
|
||||||
if len(batch) == 0 && len(peersDown) == 0 {
|
if len(batch) == 0 && len(peersDown) == 0 {
|
||||||
return false
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
startTime := time.Now()
|
startTime := time.Now()
|
||||||
@@ -441,8 +440,6 @@ func (h *PrefixHandler) flushBatch() bool {
|
|||||||
"success", successCount,
|
"success", successCount,
|
||||||
"duration_ms", elapsed.Milliseconds(),
|
"duration_ms", elapsed.Milliseconds(),
|
||||||
)
|
)
|
||||||
|
|
||||||
return true
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// applyToTable applies update to the routing table in memory: as an
|
// applyToTable applies update to the routing table in memory: as an
|
||||||
|
|||||||
Reference in New Issue
Block a user