Serve a cache hit with no database read or write (closes #309)
check / check (push) Waiting to run
check / check (push) Waiting to run
A hit opens the variant file and nothing else: its content type comes from the file's first bytes, and a file in no output format is a miss. Its access time and hit count, and the access time of a cached source a miss uses, go without waiting to the goroutine that wrote late counts, which now writes them all once a second; each kind of access time waits in a channel of 10,000 that drops the rest. checkNegativeCache leaves expired rows to CleanExpired. The .meta files, metaCache, Lookup and golang-lru go, with their tests. Dropping the .meta skip made the variant and source reconciliation walks identical, so they share one. make bench runs a parallel cache hit benchmark on the host, outside make check. Model: opus-5-5
This commit is contained in:
@@ -2,17 +2,26 @@ package imgcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
// StartPendingCountWrites starts the goroutine that writes the pending
|
||||
// counts to the database whenever there are some, one UPDATE at a time:
|
||||
// however many requests pass their deadline before their counts are
|
||||
// written, at most this one write waits for the database. It is a no-op
|
||||
// when already started. The goroutine outlives the caller, so it runs with
|
||||
// its own context, which StopPendingCountWrites cancels to tell it to
|
||||
// finish.
|
||||
// pendingWriteInterval is how often the goroutine StartPendingCountWrites
|
||||
// starts writes the pending access times and counts. SQLite's
|
||||
// CURRENT_TIMESTAMP, which an access time is written as, counts whole
|
||||
// seconds.
|
||||
const pendingWriteInterval = time.Second
|
||||
|
||||
// StartPendingCountWrites starts the goroutine that writes, every
|
||||
// pendingWriteInterval, the pending access times in one transaction and then
|
||||
// the pending counts in one UPDATE. Requests only hand these to it, so none
|
||||
// waits for the database to record what it did, and only this goroutine's
|
||||
// write ever waits for it. It is a no-op when already started. The goroutine
|
||||
// outlives the caller, so it runs with its own context, which
|
||||
// StopPendingCountWrites cancels to tell it to finish.
|
||||
func (c *Cache) StartPendingCountWrites() {
|
||||
if c.pendingCountsCancel != nil {
|
||||
return
|
||||
@@ -28,12 +37,12 @@ func (c *Cache) StartPendingCountWrites() {
|
||||
|
||||
// StopPendingCountWrites tells the goroutine StartPendingCountWrites
|
||||
// started, if it did, to finish, and waits for its write under way to end
|
||||
// and for it to return. It then writes the pending counts left, so that
|
||||
// they are written before the database closes. It waits for all of this at
|
||||
// most until ctx ends, then logs the counts not written and returns an
|
||||
// error.
|
||||
// and for it to return. It then writes the pending access times and counts
|
||||
// left, once, so that they are written before the database closes. It waits
|
||||
// for all of this at most until ctx ends. It logs what was not written,
|
||||
// access times or counts, and returns an error.
|
||||
func (c *Cache) StopPendingCountWrites(ctx context.Context) error {
|
||||
var err error
|
||||
var accessTimesErr, countsErr error
|
||||
|
||||
if c.pendingCountsCancel != nil {
|
||||
c.pendingCountsCancel()
|
||||
@@ -41,71 +50,134 @@ func (c *Cache) StopPendingCountWrites(ctx context.Context) error {
|
||||
select {
|
||||
case <-c.pendingCountsDone:
|
||||
case <-ctx.Done():
|
||||
// The write given up on here may not reach the database:
|
||||
// database/sql closes without waiting for a query, and a
|
||||
// write still waiting for the connection fails when the
|
||||
// database closes.
|
||||
err = fmt.Errorf("pending counts still being written: %w", ctx.Err())
|
||||
// The write given up on here, of access times or of counts,
|
||||
// may not reach the database: database/sql closes without
|
||||
// waiting for a query, and a write still waiting for the
|
||||
// connection fails when the database closes. Neither is
|
||||
// written here.
|
||||
accessTimesErr = fmt.Errorf("access times still being written: %w", ctx.Err())
|
||||
countsErr = fmt.Errorf("pending counts still being written: %w", ctx.Err())
|
||||
}
|
||||
}
|
||||
|
||||
if err == nil {
|
||||
err = c.writePendingCounts(ctx)
|
||||
if accessTimesErr == nil && countsErr == nil {
|
||||
accessTimesErr = c.writePendingAccessTimes(ctx)
|
||||
countsErr = c.writePendingCounts(ctx)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
if accessTimesErr != nil {
|
||||
c.log.Error("access times not written at shutdown", "error", accessTimesErr)
|
||||
}
|
||||
|
||||
if countsErr != nil {
|
||||
c.log.Error("counts not written at shutdown",
|
||||
"hits", c.pendingHits.Load(),
|
||||
"misses", c.pendingMisses.Load(),
|
||||
"upstream_fetches", c.pendingUpstreamFetches.Load(),
|
||||
"upstream_fetch_bytes", c.pendingUpstreamFetchBytes.Load(),
|
||||
"transforms", c.pendingTransforms.Load(),
|
||||
"error", err,
|
||||
"error", countsErr,
|
||||
)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
return errors.Join(accessTimesErr, countsErr)
|
||||
}
|
||||
|
||||
// pendingCountWriteLoop is the body of the goroutine
|
||||
// StartPendingCountWrites starts. Once ctx is cancelled it starts no write
|
||||
// and returns, but does not cut off a write under way then. A write that
|
||||
// fails leaves the counts pending, for the next write.
|
||||
// and returns, but does not cut off a write under way then. Counts whose
|
||||
// write fails stay pending, for the next write; access times whose write
|
||||
// fails are dropped.
|
||||
func (c *Cache) pendingCountWriteLoop(ctx context.Context) {
|
||||
defer close(c.pendingCountsDone)
|
||||
|
||||
ticker := time.NewTicker(pendingWriteInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-c.pendingCountsAdded:
|
||||
case <-ticker.C:
|
||||
}
|
||||
|
||||
// select picks at random when ctx is cancelled and a wakeup
|
||||
// waits too.
|
||||
// select picks at random when ctx is cancelled and a tick waits
|
||||
// too.
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err := c.writePendingCounts(context.WithoutCancel(ctx))
|
||||
err := c.writePendingAccessTimes(context.WithoutCancel(ctx))
|
||||
if err != nil {
|
||||
c.log.Warn("failed to write access times", "error", err)
|
||||
}
|
||||
|
||||
err = c.writePendingCounts(context.WithoutCancel(ctx))
|
||||
if err != nil {
|
||||
c.log.Warn("failed to write pending counts", "error", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// addPendingCount adds n to pendingCount, one of the pending counts, and
|
||||
// wakes the goroutine that writes them. pendingCountsAdded has capacity one,
|
||||
// so a wakeup already waiting is enough.
|
||||
func (c *Cache) addPendingCount(pendingCount *atomic.Int64, n int64) {
|
||||
pendingCount.Add(n)
|
||||
// writePendingAccessTimes takes out the variants waiting in
|
||||
// pendingVariantAccessTimes and the sources waiting in
|
||||
// pendingSourceAccessTimes, and sets their last_accessed_at to now in one
|
||||
// transaction, once for each however often it was used. Nothing else takes
|
||||
// from those channels while it runs: only the goroutine
|
||||
// StartPendingCountWrites starts calls it, or StopPendingCountWrites once
|
||||
// that goroutine has returned.
|
||||
func (c *Cache) writePendingAccessTimes(ctx context.Context) error {
|
||||
variants := make(map[imagerequest.VariantKey]struct{})
|
||||
|
||||
select {
|
||||
case c.pendingCountsAdded <- struct{}{}:
|
||||
default:
|
||||
for range len(c.pendingVariantAccessTimes) {
|
||||
variants[<-c.pendingVariantAccessTimes] = struct{}{}
|
||||
}
|
||||
|
||||
sources := make(map[imagerequest.SourceKey]struct{})
|
||||
|
||||
for range len(c.pendingSourceAccessTimes) {
|
||||
sources[<-c.pendingSourceAccessTimes] = struct{}{}
|
||||
}
|
||||
|
||||
accessTimes := len(variants) + len(sources)
|
||||
if accessTimes == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
tx, err := c.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
|
||||
// Does nothing once the transaction is committed
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
|
||||
for cacheKey := range variants {
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
UPDATE variant_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE cache_key = ?
|
||||
`, string(cacheKey))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
}
|
||||
|
||||
for sourceKey := range sources {
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE source_key = ?
|
||||
`, string(sourceKey))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
}
|
||||
|
||||
err = tx.Commit()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// writePendingCounts adds the pending counts to the cache_stats row in one
|
||||
|
||||
Reference in New Issue
Block a user