diff --git a/README.md b/README.md index 84a7bce..e5cd97a 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,7 @@ routewatch/ - Signal handlers registered for graceful shutdown 2. **Initialization** (`internal/routewatch/app.go`) - - Database created with SQLite schema (WAL mode, 3GB cache) + - Database created with SQLite schema (WAL mode, page caches as in "Memory") - Message handlers registered with the streamer - HTTP server started on configured port @@ -215,8 +215,12 @@ data it tracks. The image sets ceilings that keep it inside a 5 GiB container. Budget: - Go heap: a 1.5 GiB soft limit (`GOMEMLIMIT=1536MiB`, set in the image). -- SQLite: at most 640 MiB of page cache across the connection pool (64 MiB per - connection, 10 connections) and a 1.5 GiB hard heap limit for the C library. +- SQLite: 1 GiB of page cache for the one connection every write uses, which + keeps the pages the route writes touch from one batch to the next, and 16 MiB + for each of the 9 reading connections, whose cache SQLite empties whenever + another connection has written: about 1.15 GiB in all. A 1.5 GiB soft heap + limit sits above the caches, so it does not shrink them, and a 2 GiB hard heap + limit caps the C library. - glibc allocator: the SQLite C library runs on glibc `malloc`, which frees page-cache chunks back to per-arena free lists rather than to the kernel, so process RSS tracks the high-water mark of those arenas, not SQLite's live @@ -226,6 +230,9 @@ Budget: database writes are already serialized. - About 0.2 GiB for everything else in the runtime. +The ceilings add up to 3.7 GiB: the Go heap's 1.5 GiB, SQLite's 2 GiB and the +0.2 GiB for the rest. + Run the container with a memory limit of 5 GiB and swap disabled: ```bash @@ -244,8 +251,8 @@ What happens at each limit: - Go soft limit: as the heap approaches `GOMEMLIMIT`, the runtime runs garbage collection more aggressively rather than growing further. -- SQLite: at a 1 GiB soft heap limit it recycles its page cache instead of - allocating more; at the 1.5 GiB hard heap limit a statement fails with an +- SQLite: at the 1.5 GiB soft heap limit it recycles its page cache instead of + allocating more; at the 2 GiB hard heap limit a statement fails with an out-of-memory error, and the handler logs it and drops that batch. The process keeps running. - Handler queues: each of the four handler queues holds at most 20,000 messages. diff --git a/internal/database/database.go b/internal/database/database.go index 08ac7d1..30c4722 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -1,18 +1,21 @@ package database import ( + "cmp" "context" "database/sql" _ "embed" "encoding/json" "errors" "fmt" + "maps" "math/rand/v2" "net" "net/netip" "os" "path/filepath" "runtime" + "slices" "sync" "time" @@ -51,20 +54,31 @@ const ( minLookupMaskLengthV6 = 16 ) -// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every -// pooled connection gets them; the heap limits are process-wide and set once. +// SQLite memory tuning (README "Memory"). cache_size and busy_timeout go in the +// DSN so every pooled connection gets them; the connection kept for writing +// then sets its own cache size. The heap limits are process-wide and set once. const ( - // sqliteCacheSizeKiB is the per-connection page cache; negative means KiB. - // -65536 = 64 MiB, so at most 640 MiB across the 10-connection pool. - sqliteCacheSizeKiB = -65536 + // maxConns is the size of the connection pool: the connection kept for + // writing and 9 reading connections. + maxConns = 10 + // sqliteReadCacheSizeKiB is the page cache of each reading connection; + // negative means KiB. -16384 = 16 MiB, at most 144 MiB for the 9. A reading + // connection's cache is emptied whenever another connection has written + // since its last read, many times a minute, so a large one is wasted. + sqliteReadCacheSizeKiB = -16384 + // sqliteWriteCacheSizeKiB is the page cache of the connection kept for + // writing, -1048576 = 1 GiB. It keeps its pages from one write to the next, + // so the index pages the route writes touch stay in memory. + sqliteWriteCacheSizeKiB = -1048576 // sqliteBusyTimeoutMs is how long a connection waits on a locked database. sqliteBusyTimeoutMs = 5000 - // sqliteSoftHeapLimitBytes (1 GiB) makes SQLite recycle its cache rather - // than allocate once its C heap passes this size. - sqliteSoftHeapLimitBytes = 1073741824 - // sqliteHardHeapLimitBytes (1.5 GiB) fails a statement with SQLITE_NOMEM + // sqliteSoftHeapLimitBytes (1.5 GiB) makes SQLite recycle its cache rather + // than allocate once its C heap passes this size. It is above the caches + // together, so it does not shrink them. + sqliteSoftHeapLimitBytes = 1610612736 + // sqliteHardHeapLimitBytes (2 GiB) fails a statement with SQLITE_NOMEM // instead of growing the C heap without bound. - sqliteHardHeapLimitBytes = 1610612736 + sqliteHardHeapLimitBytes = 2147483648 ) // Settings of the ANALYZE that Analyze runs. @@ -105,7 +119,10 @@ var ( // Database manages the SQLite database connection and operations. type Database struct { - db *sql.DB + db *sql.DB + // writer is the connection kept for writing, taken from db when the + // database opens. Every write uses it while holding mu. + writer *sql.Conn logger *logger.Logger path string mu sync.Mutex @@ -143,7 +160,7 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { "file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d"+ "&_auto_vacuum=incremental&_journal_mode=WAL&_txlock=immediate", dbPath, - sqliteCacheSizeKiB, + sqliteReadCacheSizeKiB, sqliteBusyTimeoutMs, ) @@ -152,20 +169,34 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { return nil, fmt.Errorf("failed to open database: %w", err) } - err = db.PingContext(context.Background()) + ctx := context.Background() + + err = db.PingContext(ctx) if err != nil { return nil, fmt.Errorf("failed to ping database: %w", err) } // Set connection pool parameters // Multiple connections allow concurrent reads while writes are serialized - const maxConns = 10 db.SetMaxOpenConns(maxConns) db.SetMaxIdleConns(maxConns) db.SetConnMaxLifetime(0) + writer, err := db.Conn(ctx) + if err != nil { + return nil, fmt.Errorf("failed to open the writing connection: %w", err) + } + + _, err = writer.ExecContext(ctx, + fmt.Sprintf("PRAGMA cache_size=%d", sqliteWriteCacheSizeKiB)) + if err != nil { + return nil, fmt.Errorf("failed to set the writing connection's cache: %w", + err) + } + database := &Database{ - db: db, logger: logger, path: dbPath, counts: &liveCounts{}, + db: db, writer: writer, logger: logger, path: dbPath, + counts: &liveCounts{}, } err = database.Initialize() @@ -187,10 +218,11 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { // Initialize creates the database schema if it doesn't exist. func (d *Database) Initialize() error { // Set SQLite pragmas for performance. Per-connection settings (cache_size, - // synchronous, busy_timeout, journal_mode) live in the DSN; temp_store is - // left at its default so DISTINCT temp B-trees spill to disk instead of C - // heap. The heap limits below are process-wide, so setting them once here is - // enough for the whole pool. + // synchronous, busy_timeout, journal_mode) live in the DSN, and New sets the + // writing connection's cache_size; temp_store is left at its default so + // DISTINCT temp B-trees spill to disk instead of C heap. The heap limits + // below are process-wide, so setting them once here is enough for the whole + // pool. pragmas := []string{ "PRAGMA journal_mode=WAL", // Write-Ahead Logging fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes), @@ -207,7 +239,7 @@ func (d *Database) Initialize() error { // Run WAL checkpoint on startup to consolidate any existing WAL var walPages, checkpointedPages, movedPages int - err := d.db.QueryRowContext(context.Background(), + err := d.writer.QueryRowContext(context.Background(), "PRAGMA wal_checkpoint(TRUNCATE)", ).Scan(&walPages, &checkpointedPages, &movedPages) if err != nil { @@ -228,9 +260,9 @@ func (d *Database) Initialize() error { return nil } -// Close closes the database connection. +// Close closes the writing connection and then the connection pool. func (d *Database) Close() error { - return d.db.Close() + return errors.Join(d.writer.Close(), d.db.Close()) } // A live-route upsert sets the origin AS of the peer's route for the prefix @@ -428,9 +460,18 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error { } defer d.rollback(tx) + // The rows are written in key order, peer then prefix, so consecutive writes + // land on the same index pages. The sort is stable, so of two updates of + // one route the later one is still written last. + sorted := slices.SortedStableFunc(slices.Values(routes), + func(a, b *LiveRoute) int { + return cmp.Or(cmp.Compare(a.PeerIP, b.PeerIP), + cmp.Compare(a.Prefix, b.Prefix)) + }) + var routesV4, routesV6 []*LiveRoute - for _, route := range routes { + for _, route := range sorted { if route.IPVersion == ipVersionV4 { routesV4 = append(routesV4, route) } else { @@ -574,9 +615,17 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error { } defer d.rollback(tx) + // The rows are deleted in key order, peer then prefix, as + // UpsertLiveRouteBatch writes them. + sorted := slices.SortedFunc(slices.Values(deletions), + func(a, b LiveRouteDeletion) int { + return cmp.Or(cmp.Compare(a.PeerIP, b.PeerIP), + cmp.Compare(a.Prefix, b.Prefix)) + }) + var deletionsV4, deletionsV6 []LiveRouteDeletion - for _, del := range deletions { + for _, del := range sorted { if del.IPVersion == ipVersionV4 { deletionsV4 = append(deletionsV4, del) } else { @@ -780,9 +829,10 @@ func upsertPrefixesInTable( var newPrefixes int - for prefix, timestamp := range prefixes { + // The rows are written in key order, as UpsertLiveRouteBatch writes routes. + for _, prefix := range slices.Sorted(maps.Keys(prefixes)) { inserted, err := upsertPrefix(ctx, updateStmt, insertStmt, prefix, - timestamp) + prefixes[prefix]) if err != nil { return 0, err } @@ -1411,25 +1461,25 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error { // The write lock is held, so no other writer can insert this key between the // update and the insert even though they are separate autocommit statements. - setOrigin, err := d.db.PrepareContext(ctx, originSQL) + setOrigin, err := d.writer.PrepareContext(ctx, originSQL) if err != nil { return fmt.Errorf("failed to prepare origin statement: %w", err) } defer func() { _ = setOrigin.Close() }() - upd, err := d.db.PrepareContext(ctx, updateSQL) + upd, err := d.writer.PrepareContext(ctx, updateSQL) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } defer func() { _ = upd.Close() }() - ins, err := d.db.PrepareContext(ctx, insertSQL) + ins, err := d.writer.PrepareContext(ctx, insertSQL) if err != nil { return fmt.Errorf("failed to prepare insert statement: %w", err) } defer func() { _ = ins.Close() }() - has, err := d.db.PrepareContext(ctx, lookupSQL) + has, err := d.writer.PrepareContext(ctx, lookupSQL) if err != nil { return fmt.Errorf("failed to prepare prefix lookup statement: %w", err) } @@ -2025,7 +2075,7 @@ func (d *Database) UpdateASNWHOIS( WHERE asn = ? ` - _, err := d.db.ExecContext(ctx, query, + _, err := d.writer.ExecContext(ctx, query, update.ASName, update.OrgName, update.OrgID, @@ -2191,7 +2241,7 @@ func (d *Database) Vacuum(ctx context.Context) error { d.lock("Vacuum") defer d.unlock() - rows, err := d.db.QueryContext(ctx, + rows, err := d.writer.QueryContext(ctx, fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree)) if err != nil { return fmt.Errorf("failed to run incremental vacuum: %w", err) @@ -2223,7 +2273,7 @@ func (d *Database) Analyze(ctx context.Context) error { // analysis_limit is a setting of the connection, so it is set in the same // call as PRAGMA optimize, which runs both statements on one connection. - _, err := d.db.ExecContext(ctx, fmt.Sprintf( + _, err := d.writer.ExecContext(ctx, fmt.Sprintf( "PRAGMA analysis_limit=%d; PRAGMA optimize=%#x", analysisLimit, optimizeAllTables)) if err != nil { @@ -2240,7 +2290,7 @@ func (d *Database) Checkpoint(ctx context.Context) error { d.lock("Checkpoint") defer d.unlock() - _, err := d.db.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)") + _, err := d.writer.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)") if err != nil { return fmt.Errorf("failed to checkpoint WAL: %w", err) } @@ -2296,9 +2346,9 @@ func (d *Database) unlock() { "held_by", lockedBy, "duration_ms", holdDuration.Milliseconds()) } -// beginTx starts a new transaction with logging +// beginTx starts a new transaction on the writing connection, with logging func (d *Database) beginTx() (*loggingTx, error) { - tx, err := d.db.BeginTx(context.Background(), nil) + tx, err := d.writer.BeginTx(context.Background(), nil) if err != nil { return nil, err } diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index 2be88da..639e9fb 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -9,6 +9,7 @@ import ( "math/rand/v2" "net/netip" "slices" + "strconv" "sync" "testing" "time" @@ -33,6 +34,9 @@ const vacuumTestRoutes = 2000 // once so each is a distinct SQLite connection that parsed the DSN. const heldConnections = 5 +// bytesPerKiB turns a cache_size in KiB into bytes. +const bytesPerKiB = 1024 + // testPeerIP is the peer every route in the IP lookup test is learned from. const testPeerIP = "192.0.2.254" @@ -336,8 +340,10 @@ func writeOriginRoutes(t *testing.T, db *Database, origins map[string]int) { // TestConnectionPoolPragmas holds several pooled connections open at once and // checks each one carries the per-connection settings from the DSN, plus the -// process-wide hard heap limit, and sees the new file with auto_vacuum -// incremental. +// process-wide heap limits, and sees the new file with auto_vacuum +// incremental. The connection kept for writing must have the large cache, and +// the soft heap limit must leave room for every connection's cache: SQLite +// shrinks the caches to stay under it. func TestConnectionPoolPragmas(t *testing.T) { t.Parallel() @@ -371,9 +377,9 @@ func TestConnectionPoolPragmas(t *testing.T) { for i, c := range conns { cacheSize := readPragma(ctx, t, c, i, "cache_size") - if cacheSize != sqliteCacheSizeKiB { + if cacheSize != sqliteReadCacheSizeKiB { t.Errorf("conn %d: cache_size = %d, want %d", - i, cacheSize, sqliteCacheSizeKiB) + i, cacheSize, sqliteReadCacheSizeKiB) } busyTimeout := readPragma(ctx, t, c, i, "busy_timeout") @@ -388,6 +394,12 @@ func TestConnectionPoolPragmas(t *testing.T) { i, tempStore, tempStoreMemory) } + softHeapLimit := readPragma(ctx, t, c, i, "soft_heap_limit") + if softHeapLimit != sqliteSoftHeapLimitBytes { + t.Errorf("conn %d: soft_heap_limit = %d, want %d", + i, softHeapLimit, sqliteSoftHeapLimitBytes) + } + hardHeapLimit := readPragma(ctx, t, c, i, "hard_heap_limit") if hardHeapLimit != sqliteHardHeapLimitBytes { t.Errorf("conn %d: hard_heap_limit = %d, want %d", @@ -400,6 +412,29 @@ func TestConnectionPoolPragmas(t *testing.T) { i, autoVacuum, autoVacuumIncremental) } } + + var writeCacheSize int64 + + err = db.writer.QueryRowContext(ctx, "PRAGMA cache_size"). + Scan(&writeCacheSize) + if err != nil { + t.Fatalf("failed to read the writing connection's cache_size: %v", err) + } + + if writeCacheSize != sqliteWriteCacheSizeKiB { + t.Errorf("writing connection: cache_size = %d, want %d", + writeCacheSize, sqliteWriteCacheSizeKiB) + } + + readCacheSize := readPragma(ctx, t, conns[0], 0, "cache_size") + softHeapLimit := readPragma(ctx, t, conns[0], 0, "soft_heap_limit") + cachesBytes := -bytesPerKiB * + (writeCacheSize + (maxConns-1)*readCacheSize) + + if softHeapLimit <= cachesBytes { + t.Errorf("soft_heap_limit = %d, not above the %d bytes of the caches", + softHeapLimit, cachesBytes) + } } // readPragma reads the integer value of PRAGMA name on connection i, c. @@ -1229,6 +1264,85 @@ func TestUpdatePrefixesBatchKeepsFirstSeen(t *testing.T) { } } +// TestUpsertLiveRouteBatchWritesInKeyOrder writes new routes of two peers out +// of key order, the last one a second announcement of the first, and checks +// that the rows were inserted in key order, peer then prefix (each new row's id +// is one above the last), and that the route announced twice carries the later +// announcement. +func TestUpsertLiveRouteBatchWritesInKeyOrder(t *testing.T) { + t.Parallel() + + db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + const firstOrigin, laterOrigin = 64500, 64501 + + type row struct { + peer, prefix string + origin int + } + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + + var routes []*LiveRoute + + for _, announced := range []row{ + {secondV4PeerIP, "198.51.100.0/24", firstOrigin}, + {testV4PeerIP, "203.0.113.0/24", firstOrigin}, + {secondV4PeerIP, "192.0.2.0/24", firstOrigin}, + {testV4PeerIP, "198.51.100.0/24", firstOrigin}, + {secondV4PeerIP, "198.51.100.0/24", laterOrigin}, + } { + route := mkV4Route(t, announced.prefix, announced.origin, ts) + route.PeerIP = announced.peer + routes = append(routes, route) + } + + err = db.UpsertLiveRouteBatch(routes) + if err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + rows, err := db.db.QueryContext(t.Context(), + "SELECT peer_ip, prefix, origin_asn FROM live_routes_v4 ORDER BY id") + if err != nil { + t.Fatalf("reading the routes: %v", err) + } + defer func() { _ = rows.Close() }() + + var stored []row + + for rows.Next() { + var r row + + err := rows.Scan(&r.peer, &r.prefix, &r.origin) + if err != nil { + t.Fatalf("reading a route: %v", err) + } + + stored = append(stored, r) + } + + err = rows.Err() + if err != nil { + t.Fatalf("reading the routes: %v", err) + } + + want := []row{ + {testV4PeerIP, "198.51.100.0/24", firstOrigin}, + {testV4PeerIP, "203.0.113.0/24", firstOrigin}, + {secondV4PeerIP, "192.0.2.0/24", firstOrigin}, + {secondV4PeerIP, "198.51.100.0/24", laterOrigin}, + } + + if !slices.Equal(stored, want) { + t.Errorf("routes in insertion order = %+v, want %+v", stored, want) + } +} + // Parameters for BenchmarkPrefixBatchOnFiveMillionRoutes. const ( // benchPeers and benchPrefixes size the table the batches are written to: @@ -1238,6 +1352,10 @@ const ( // benchBatchSize is how many updates each batch holds, as many as a full // prefix handler batch (prefixBatchSize in internal/routewatch). benchBatchSize = 25000 + // benchMaxBatchSize is the largest batch the prefix handler writes, the + // one it holds while a slow write runs (prefixBatchMaxSize in + // internal/routewatch). + benchMaxBatchSize = 250000 // benchWithdrawalEvery makes every fifth update of a timed batch a // withdrawal; the others are announcements. benchWithdrawalEvery = 5 @@ -1250,13 +1368,14 @@ const ( ) // BenchmarkPrefixBatchOnFiveMillionRoutes times the writes the prefix handler -// makes for one full batch, 25,000 announcements and withdrawals of routes -// picked at random, to a database that already holds 5 million live routes: -// 500,000 IPv4 prefixes, each from 10 peers. It builds that database first, -// one batch at a time, which takes some minutes and a few GB of disk under -// TMPDIR. It is run by hand, not by make test, for example in the test image: +// makes for one batch of announcements and withdrawals of routes picked at +// random, to a database that already holds 5 million live routes: 500,000 +// IPv4 prefixes, each from 10 peers. It times a full batch of 25,000 updates, +// then the largest batch, 250,000. It builds that database first, one batch at +// a time, which takes some minutes and a few GB of disk under TMPDIR. It is +// run by hand, not by make test, for example in the test image: // -// go test -run '^$' -bench FiveMillion -benchtime 20x ./internal/database/ +// go test -run '^$' -bench FiveMillion -benchtime 10x ./internal/database/ func BenchmarkPrefixBatchOnFiveMillionRoutes(b *testing.B) { db, err := New(&config.Config{StateDir: b.TempDir()}, logger.New()) if err != nil { @@ -1280,28 +1399,32 @@ func BenchmarkPrefixBatchOnFiveMillionRoutes(b *testing.B) { } } - for b.Loop() { - ts = ts.Add(time.Second) + for _, batchSize := range []int{benchBatchSize, benchMaxBatchSize} { + b.Run(strconv.Itoa(batchSize), func(b *testing.B) { + for b.Loop() { + ts = ts.Add(time.Second) - var ( - routes []*LiveRoute - withdrawals []LiveRouteDeletion - ) + var ( + routes []*LiveRoute + withdrawals []LiveRouteDeletion + ) - for i := range benchBatchSize { - route := benchRoute(rng, rng.IntN(benchPrefixes), - rng.IntN(benchPeers), ts) - if i%benchWithdrawalEvery == 0 { - withdrawals = append(withdrawals, LiveRouteDeletion{ - Prefix: route.Prefix, PeerIP: route.PeerIP, - IPVersion: ipVersionV4, - }) - } else { - routes = append(routes, route) + for i := range batchSize { + route := benchRoute(rng, rng.IntN(benchPrefixes), + rng.IntN(benchPeers), ts) + if i%benchWithdrawalEvery == 0 { + withdrawals = append(withdrawals, LiveRouteDeletion{ + Prefix: route.Prefix, PeerIP: route.PeerIP, + IPVersion: ipVersionV4, + }) + } else { + routes = append(routes, route) + } + } + + writeBenchBatch(b, db, routes, withdrawals, ts) } - } - - writeBenchBatch(b, db, routes, withdrawals, ts) + }) } } diff --git a/internal/database/slowquery.go b/internal/database/slowquery.go index d027211..13a0ac8 100644 --- a/internal/database/slowquery.go +++ b/internal/database/slowquery.go @@ -18,12 +18,12 @@ func logSlowQuery(logger *logger.Logger, query string, start time.Time) { } } -// exec wraps Exec with slow query logging +// exec runs a statement on the writing connection, with slow query logging func (d *Database) exec(query string, args ...any) error { start := time.Now() defer logSlowQuery(d.logger, query, start) - _, err := d.db.ExecContext(context.Background(), query, args...) + _, err := d.writer.ExecContext(context.Background(), query, args...) return err }