next into main #166

Merged
sneak merged 24 commits from next into main 2026-10-27 18:15:11 +01:00
6 changed files with 78 additions and 30 deletions
Showing only changes of commit 2c304a0022 - Show all commits
+4
View File
@@ -33,6 +33,10 @@ tested.
# 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
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
+3 -5
View File
@@ -202,11 +202,11 @@ func (h *ASHandler) takeBatch() []asnOp {
}
// 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.
func (h *ASHandler) flushBatch() bool {
// holding the mutex during the write.
func (h *ASHandler) flushBatch() {
batch := h.takeBatch()
if len(batch) == 0 {
return false
return
}
// 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.writing = 0
h.mu.Unlock()
return true
}
+13 -15
View File
@@ -6,27 +6,26 @@ import (
)
// 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,
// writes it, and reports whether there was one.
// the only goroutine that writes a handler's batch. flushBatch takes the batch
// and writes it.
//
// flushLoop calls flushBatch when batchFullCh wakes it. It also calls it every
// batchTimeout, unless it wrote a batch less than batchTimeout ago. When stopCh
// hands it Stop's context, it calls flushBatch once more, unless the stop
// deadline has passed. Then it closes done and returns.
// flushLoop calls flushBatch on each tick of a timer, every batchTimeout. It
// also calls it at once when batchFullCh wakes it. A full batch's write
// restarts the timer, so the next tick comes batchTimeout after that write
// 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(
batchTimeout time.Duration,
batchFullCh <-chan struct{},
stopCh <-chan context.Context,
done chan<- struct{},
flushBatch func() bool,
flushBatch func(),
) {
defer close(done)
ticker := time.NewTicker(batchTimeout)
defer ticker.Stop()
lastFlush := time.Now()
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
@@ -45,13 +44,12 @@ func flushLoop(
select {
case <-ticker.C:
if time.Since(lastFlush) >= batchTimeout && flushBatch() {
lastFlush = time.Now()
}
flushBatch()
case <-batchFullCh:
if flushBatch() {
lastFlush = time.Now()
}
// Reset also drops a tick that waits in ticker.C, so no timed
// write follows this one at once.
ticker.Reset(batchTimeout)
flushBatch()
case ctx := <-stopCh:
// Final flush, unless the stop deadline has passed
if ctx.Err() == nil {
@@ -9,6 +9,7 @@ import (
"slices"
"sync"
"testing"
"testing/synctest"
"time"
"git.eeqj.de/sneak/routewatch/internal/database"
@@ -27,6 +28,58 @@ const (
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
// 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
+3 -5
View File
@@ -200,11 +200,11 @@ func (h *PeerHandler) takeBatch() []peerUpdate {
}
// 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.
func (h *PeerHandler) flushBatch() bool {
// holding the mutex during the write.
func (h *PeerHandler) flushBatch() {
batch := h.takeBatch()
if len(batch) == 0 {
return false
return
}
// 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.writing = 0
h.mu.Unlock()
return true
}
+2 -5
View File
@@ -352,13 +352,12 @@ func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) {
// 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.
// It reports whether there was a batch, or a peer down to delete the routes of.
func (h *PrefixHandler) flushBatch() bool {
func (h *PrefixHandler) flushBatch() {
h.writeTableChanges()
batch, peersDown := h.takeBatch()
if len(batch) == 0 && len(peersDown) == 0 {
return false
return
}
startTime := time.Now()
@@ -441,8 +440,6 @@ func (h *PrefixHandler) flushBatch() bool {
"success", successCount,
"duration_ms", elapsed.Milliseconds(),
)
return true
}
// applyToTable applies update to the routing table in memory: as an