|
|
|
@@ -18,6 +18,14 @@ import (
|
|
|
|
|
// write-pressure wakeups triggered by stores.
|
|
|
|
|
const DefaultEvictionInterval = 5 * time.Minute
|
|
|
|
|
|
|
|
|
|
// reconciliationWaitFactor is how many times as long as a
|
|
|
|
|
// reconciliation pass took the eviction loop waits at least, after that
|
|
|
|
|
// pass ends, before it starts the next one. A pass reads a row and stats a
|
|
|
|
|
// file for every cached file, so its time grows with the cache; waiting
|
|
|
|
|
// ten times as long keeps reconciliation to at most a tenth of the time,
|
|
|
|
|
// however large the cache is.
|
|
|
|
|
const reconciliationWaitFactor = 10
|
|
|
|
|
|
|
|
|
|
// evictionBatchSize is how many LRU candidates of each class (variants
|
|
|
|
|
// and source blobs) one eviction pass fetches from the database.
|
|
|
|
|
const evictionBatchSize = 100
|
|
|
|
@@ -438,10 +446,12 @@ func (c *Cache) notifyWritePressure() {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// StartEviction launches the background eviction goroutine, which
|
|
|
|
|
// reconciles the database accounting with the cache directories and
|
|
|
|
|
// removes the expired negative cache rows at startup and again on every
|
|
|
|
|
// periodic tick thereafter, and evicts to the configured limit on the
|
|
|
|
|
// given periodic interval and on write-pressure notifications. On a
|
|
|
|
|
// periodic tick thereafter, reconciles the database accounting with the
|
|
|
|
|
// cache directories at startup and again on the first periodic tick
|
|
|
|
|
// that comes at least ten times as long after a reconciliation pass
|
|
|
|
|
// ended as that pass took, and evicts to the configured limit on the given
|
|
|
|
|
// periodic interval and on write-pressure notifications. On a
|
|
|
|
|
// disabled cache it only removes the expired negative cache rows, as
|
|
|
|
|
// the negative cache is used either way. It is a no-op when already
|
|
|
|
|
// started. The goroutine outlives the caller, so it runs with its own
|
|
|
|
@@ -482,7 +492,7 @@ func (c *Cache) StopEviction(ctx context.Context) error {
|
|
|
|
|
func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
|
|
|
|
|
defer close(c.evictionDone)
|
|
|
|
|
|
|
|
|
|
c.runReconciliationPass(ctx)
|
|
|
|
|
nextReconciliation := c.runReconciliationPass(ctx)
|
|
|
|
|
c.runCleanExpired(ctx)
|
|
|
|
|
c.runEvictionPass(ctx)
|
|
|
|
|
|
|
|
|
@@ -494,16 +504,20 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
|
|
|
|
|
case <-ctx.Done():
|
|
|
|
|
return
|
|
|
|
|
case <-ticker.C:
|
|
|
|
|
// Reconciliation walks the cache directories, so it only
|
|
|
|
|
// runs on the periodic ticker rather than on every
|
|
|
|
|
// write-pressure wakeup, keeping it off the per-store hot
|
|
|
|
|
// path. Reusing the eviction interval itself (rather than a
|
|
|
|
|
// separate, longer one) is a deliberate choice: it is the
|
|
|
|
|
// simplest option that still bounds how long a store's
|
|
|
|
|
// best-effort accounting insert can stay silently
|
|
|
|
|
// unaccounted for to one interval, on a process that is
|
|
|
|
|
// already running this loop regardless.
|
|
|
|
|
c.runReconciliationPass(ctx)
|
|
|
|
|
// Reconciliation walks the cache directories and reads a
|
|
|
|
|
// row for every file, so it runs only on the periodic
|
|
|
|
|
// ticker rather than on every write-pressure wakeup, and
|
|
|
|
|
// only on the first tick at or after nextReconciliation,
|
|
|
|
|
// which is ten times as long after the last pass ended as
|
|
|
|
|
// that pass took. Passes then never run back to back,
|
|
|
|
|
// however long one takes. A file whose best-effort
|
|
|
|
|
// accounting insert failed in a store stays unaccounted for
|
|
|
|
|
// until the next pass, up to about one eviction interval
|
|
|
|
|
// plus ten times the last pass's time.
|
|
|
|
|
if !time.Now().Before(nextReconciliation) {
|
|
|
|
|
nextReconciliation = c.runReconciliationPass(ctx)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
c.runCleanExpired(ctx)
|
|
|
|
|
case <-c.evictionPressure:
|
|
|
|
|
}
|
|
|
|
@@ -527,17 +541,25 @@ func (c *Cache) runEvictionPass(ctx context.Context) {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// runReconciliationPass runs one reconciliation pass, logging failures
|
|
|
|
|
// instead of propagating them (the loop must keep running). It does
|
|
|
|
|
// nothing once ctx is cancelled.
|
|
|
|
|
func (c *Cache) runReconciliationPass(ctx context.Context) {
|
|
|
|
|
// instead of propagating them (the loop must keep running), and returns
|
|
|
|
|
// the earliest time the next pass may start: reconciliationWaitFactor
|
|
|
|
|
// times as long after this pass ended as it took. It does nothing once
|
|
|
|
|
// ctx is cancelled.
|
|
|
|
|
func (c *Cache) runReconciliationPass(ctx context.Context) time.Time {
|
|
|
|
|
if ctx.Err() != nil {
|
|
|
|
|
return
|
|
|
|
|
return time.Now()
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
started := time.Now()
|
|
|
|
|
|
|
|
|
|
err := c.reconcileAccounting(ctx)
|
|
|
|
|
if err != nil {
|
|
|
|
|
c.log.Warn("cache accounting reconciliation failed", "error", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
ended := time.Now()
|
|
|
|
|
|
|
|
|
|
return ended.Add(reconciliationWaitFactor * ended.Sub(started))
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// runCleanExpired runs CleanExpired, logging a failure instead of
|
|
|
|
@@ -556,7 +578,8 @@ func (c *Cache) runCleanExpired(ctx context.Context) {
|
|
|
|
|
|
|
|
|
|
// reconcileAccounting synchronizes the database size accounting with
|
|
|
|
|
// the actual contents of the cache directories. It runs at startup and
|
|
|
|
|
// again on every periodic eviction tick thereafter, off the request
|
|
|
|
|
// then on the first periodic eviction tick that comes at least ten times
|
|
|
|
|
// as long after the last pass ended as that pass took, off the request
|
|
|
|
|
// hot path: it adopts variant files that predate the accounting table
|
|
|
|
|
// (or whose accounting insert failed, e.g. StoreVariant's best-effort
|
|
|
|
|
// insert under transient DB contention), drops accounting rows whose
|
|
|
|
@@ -566,9 +589,10 @@ func (c *Cache) runCleanExpired(ctx context.Context) {
|
|
|
|
|
// files left behind by crashed writes, and last checks the total cache
|
|
|
|
|
// usage against the tables. It reads the tables a page at a time.
|
|
|
|
|
// Running it periodically, not just once, bounds how long such drift can
|
|
|
|
|
// accumulate unaccounted for on a long-running process to one eviction
|
|
|
|
|
// interval. Once ctx is cancelled, it stops at the next file or row and
|
|
|
|
|
// returns ctx's error.
|
|
|
|
|
// accumulate unaccounted for on a long-running process to about one
|
|
|
|
|
// eviction interval plus ten times the last pass's time.
|
|
|
|
|
// Once ctx is cancelled, it stops at the next file or row and returns
|
|
|
|
|
// ctx's error.
|
|
|
|
|
func (c *Cache) reconcileAccounting(ctx context.Context) error {
|
|
|
|
|
if c.disabled {
|
|
|
|
|
return nil
|
|
|
|
|