feat(ermitegit): bound memory, close idle repositories, claim refreshes and close synchronously
- commit
- a4afa9bb7cc52a13cd6b5bf454b0758af40f6d2f
- author
- brice · · 2026-10-11 00:46 UTC
- parent
- 5925475
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +335 -41
M internal/adapters/ermitegit/cache.go +176 -33
| @@ -25,7 +25,22 @@ const DefaultMaxRepos = 64 | |||
| 25 | 25 | ||
| 26 | 26 | // DefaultRefreshEvery is how long a repository goes without a Refresh when | |
| 27 | 27 | // Config.RefreshEvery is zero. See Config.RefreshEvery for what depends on it. | |
| 28 | - | const DefaultRefreshEvery = time.Second | |
| 28 | + | // A write that goes through ermite refreshes the repository at once (see | |
| 29 | + | // Adapter.Refresh), so this is only the safety net for the other writers. | |
| 30 | + | const DefaultRefreshEvery = 5 * time.Second | |
| 31 | + | ||
| 32 | + | // DefaultIdleTTL is how long a repository may go unused before it is closed, | |
| 33 | + | // when Config.IdleTTL is zero. | |
| 34 | + | const DefaultIdleTTL = 10 * time.Minute | |
| 35 | + | ||
| 36 | + | // DefaultDeltaCacheSize is the budget of the cache of delta bases of each open | |
| 37 | + | // repository when Config.DeltaCacheSize is zero (the library's own is 32 MiB). | |
| 38 | + | const DefaultDeltaCacheSize = 8 << 20 | |
| 39 | + | ||
| 40 | + | // DefaultMaxObjectSize is the largest object a repository holds in memory when | |
| 41 | + | // Config.MaxObjectSize is zero (the library's own is 128 MiB). A bigger one | |
| 42 | + | // is read by the wrapped reader. | |
| 43 | + | const DefaultMaxObjectSize = 16 << 20 | |
| 29 | 44 | ||
| 30 | 45 | // unsupportedRetry is how long a repository that ermite-git cannot read is | |
| 31 | 46 | // served by git before Open is tried again (a repack, or the removal of a | |
| @@ -46,9 +61,10 @@ type entry struct { | |||
| 46 | 61 | ||
| 47 | 62 | refreshed atomic.Int64 // unix nanoseconds of the last Refresh | |
| 48 | 63 | ||
| 49 | - | elem *list.Element | |
| 50 | - | users int | |
| 51 | - | dropped bool // out of the cache: close when the last user is done | |
| 64 | + | elem *list.Element | |
| 65 | + | users int | |
| 66 | + | lastUsed time.Time // of the last get or release: the LRU is ordered by it | |
| 67 | + | dropped bool // out of the cache: close when the last user is done | |
| 52 | 68 | } | |
| 53 | 69 | ||
| 54 | 70 | // opened is what git.Open answered for an entry. | |
| @@ -69,25 +85,100 @@ type handle struct { | |||
| 69 | 85 | func (h *handle) release() { h.c.release(h.e) } | |
| 70 | 86 | ||
| 71 | 87 | // cache is a bounded LRU of open repositories keyed by directory. A repository | |
| 72 | - | // leaves it by eviction, by Forget (the directory was deleted) or because the | |
| 73 | - | // directory is no longer the same one (checked on each use through its | |
| 74 | - | // inode); it is closed once the reads that use it are done. | |
| 88 | + | // leaves it by eviction, by idling longer than the TTL, by Forget (the | |
| 89 | + | // directory was deleted) or because the directory is no longer the same one | |
| 90 | + | // (checked on each use through its inode); it is closed once the reads that | |
| 91 | + | // use it are done, by the goroutine that made it leave or that was the last | |
| 92 | + | // to use it. | |
| 75 | 93 | type cache struct { | |
| 76 | 94 | max int | |
| 77 | 95 | refreshEvery time.Duration | |
| 96 | + | idleTTL time.Duration // <= 0: never | |
| 97 | + | openOpts []git.Option | |
| 78 | 98 | onUnsupport func(ctx context.Context, dir string, name domain.RepoName, cause error) | |
| 79 | 99 | ||
| 80 | 100 | mu sync.Mutex | |
| 81 | 101 | entries map[string]*entry | |
| 82 | 102 | lru *list.List // front = most recently used | |
| 103 | + | doomed []*entry // dropped without users: closed by unlock | |
| 83 | 104 | closed bool | |
| 105 | + | ||
| 106 | + | stop chan struct{} // closed by close; ends the sweeper | |
| 107 | + | swept sync.WaitGroup | |
| 108 | + | stopped sync.Once | |
| 109 | + | } | |
| 110 | + | ||
| 111 | + | // cacheConfig is what newCache needs. | |
| 112 | + | type cacheConfig struct { | |
| 113 | + | max int | |
| 114 | + | refreshEvery time.Duration | |
| 115 | + | idleTTL time.Duration | |
| 116 | + | openOpts []git.Option | |
| 117 | + | onUnsupport func(context.Context, string, domain.RepoName, error) | |
| 118 | + | } | |
| 119 | + | ||
| 120 | + | func newCache(cfg cacheConfig) *cache { | |
| 121 | + | if cfg.max <= 0 { | |
| 122 | + | cfg.max = DefaultMaxRepos | |
| 123 | + | } | |
| 124 | + | c := &cache{ | |
| 125 | + | max: cfg.max, refreshEvery: cfg.refreshEvery, idleTTL: cfg.idleTTL, openOpts: cfg.openOpts, | |
| 126 | + | onUnsupport: cfg.onUnsupport, entries: make(map[string]*entry), lru: list.New(), stop: make(chan struct{}), | |
| 127 | + | } | |
| 128 | + | if c.idleTTL > 0 { | |
| 129 | + | c.swept.Add(1) | |
| 130 | + | go c.sweeper() | |
| 131 | + | } | |
| 132 | + | return c | |
| 133 | + | } | |
| 134 | + | ||
| 135 | + | // sweeper closes the repositories that idle past the TTL. It is the only | |
| 136 | + | // goroutine of the cache, whatever the number of repositories, and close ends | |
| 137 | + | // it. | |
| 138 | + | func (c *cache) sweeper() { | |
| 139 | + | defer c.swept.Done() | |
| 140 | + | tick := time.NewTicker(max(c.idleTTL/4, time.Millisecond)) | |
| 141 | + | defer tick.Stop() | |
| 142 | + | for { | |
| 143 | + | select { | |
| 144 | + | case <-c.stop: | |
| 145 | + | return | |
| 146 | + | case now := <-tick.C: | |
| 147 | + | c.sweep(now) | |
| 148 | + | } | |
| 149 | + | } | |
| 150 | + | } | |
| 151 | + | ||
| 152 | + | // sweep drops the entries nobody used since now-idleTTL. The LRU is ordered by | |
| 153 | + | // lastUsed, so the walk from the back stops at the first recent entry; one | |
| 154 | + | // that is in use is not idle and stays. | |
| 155 | + | func (c *cache) sweep(now time.Time) { | |
| 156 | + | c.mu.Lock() | |
| 157 | + | for el := c.lru.Back(); el != nil; { | |
| 158 | + | prev := el.Prev() | |
| 159 | + | e := el.Value.(*entry) | |
| 160 | + | if now.Sub(e.lastUsed) <= c.idleTTL { | |
| 161 | + | break | |
| 162 | + | } | |
| 163 | + | if e.users == 0 { | |
| 164 | + | c.dropLocked(e) | |
| 165 | + | } | |
| 166 | + | el = prev | |
| 167 | + | } | |
| 168 | + | c.unlock() | |
| 84 | 169 | } | |
| 85 | 170 | ||
| 86 | - | func newCache(max int, refreshEvery time.Duration, onUnsupport func(context.Context, string, domain.RepoName, error)) *cache { | |
| 87 | - | if max <= 0 { | |
| 88 | - | max = DefaultMaxRepos | |
| 171 | + | // unlock releases the mutex, then closes what was dropped without users while | |
| 172 | + | // it was held: in the calling goroutine, so that when close returns the | |
| 173 | + | // mappings are released, and not under the mutex, since Close takes the | |
| 174 | + | // repository's own. | |
| 175 | + | func (c *cache) unlock() { | |
| 176 | + | doomed := c.doomed | |
| 177 | + | c.doomed = nil | |
| 178 | + | c.mu.Unlock() | |
| 179 | + | for _, e := range doomed { | |
| 180 | + | e.closeRepo() | |
| 89 | 181 | } | |
| 90 | - | return &cache{max: max, refreshEvery: refreshEvery, onUnsupport: onUnsupport, entries: make(map[string]*entry), lru: list.New()} | |
| 91 | 182 | } | |
| 92 | 183 | ||
| 93 | 184 | // get returns the entry of dir, opened and refreshed, or a nil repo when | |
| @@ -108,7 +199,7 @@ func (c *cache) get(ctx context.Context, name domain.RepoName, dir string) (*han | |||
| 108 | 199 | ||
| 109 | 200 | c.mu.Lock() | |
| 110 | 201 | if c.closed { | |
| 111 | - | c.mu.Unlock() | |
| 202 | + | c.unlock() | |
| 112 | 203 | return nil, ErrClosed | |
| 113 | 204 | } | |
| 114 | 205 | e := c.entries[dir] | |
| @@ -125,7 +216,8 @@ func (c *cache) get(ctx context.Context, name domain.RepoName, dir string) (*han | |||
| 125 | 216 | c.lru.MoveToFront(e.elem) | |
| 126 | 217 | } | |
| 127 | 218 | e.users++ | |
| 128 | - | c.mu.Unlock() | |
| 219 | + | e.lastUsed = time.Now() | |
| 220 | + | c.unlock() | |
| 129 | 221 | ||
| 130 | 222 | h := &handle{c: c, e: e} | |
| 131 | 223 | e.once.Do(func() { c.open(ctx, name, e) }) | |
| @@ -151,20 +243,26 @@ func (c *cache) get(ctx context.Context, name domain.RepoName, dir string) (*han | |||
| 151 | 243 | } | |
| 152 | 244 | ||
| 153 | 245 | // refresh calls Refresh on the repository of e unless one ran less than | |
| 154 | - | // refreshEvery ago (a negative interval refreshes on every read). The library | |
| 155 | - | // reads loose refs live, stat-checks packed-refs on each ref read and rescans | |
| 156 | - | // the packs when an object is missing, so a read sees a push whether or not | |
| 157 | - | // Refresh ran; Refresh is what drops replaced packs, picks up a new | |
| 246 | + | // refreshEvery ago (a negative interval refreshes on every read). Of the reads | |
| 247 | + | // that find it due, one wins a compare-and-swap on the time of the last | |
| 248 | + | // Refresh and runs it; the others go on with the snapshot they have. The | |
| 249 | + | // library reads loose refs live, stat-checks packed-refs on each ref read and | |
| 250 | + | // rescans the packs when an object is missing, so a read sees a push whether | |
| 251 | + | // or not Refresh ran; Refresh is what drops replaced packs, picks up a new | |
| 158 | 252 | // commit-graph and notices that the repository became unsupported. | |
| 159 | 253 | func (c *cache) refresh(e *entry, repo *git.Repo) error { | |
| 254 | + | if c.refreshEvery < 0 { | |
| 255 | + | return repo.Refresh() | |
| 256 | + | } | |
| 160 | 257 | now := time.Now().UnixNano() | |
| 161 | - | if c.refreshEvery >= 0 && now-e.refreshed.Load() < int64(c.refreshEvery) { | |
| 258 | + | last := e.refreshed.Load() | |
| 259 | + | if now-last < int64(c.refreshEvery) || !e.refreshed.CompareAndSwap(last, now) { | |
| 162 | 260 | return nil | |
| 163 | 261 | } | |
| 164 | 262 | if err := repo.Refresh(); err != nil { | |
| 263 | + | e.refreshed.CompareAndSwap(now, last) // the next read tries again | |
| 165 | 264 | return err | |
| 166 | 265 | } | |
| 167 | - | e.refreshed.Store(now) | |
| 168 | 266 | return nil | |
| 169 | 267 | } | |
| 170 | 268 | ||
| @@ -178,10 +276,11 @@ func (e *entry) stale() bool { | |||
| 178 | 276 | // open runs git.Open once per entry; concurrent users wait in Once.Do. | |
| 179 | 277 | func (c *cache) open(ctx context.Context, name domain.RepoName, e *entry) { | |
| 180 | 278 | op := &opened{at: time.Now()} | |
| 181 | - | repo, err := git.Open(e.dir) | |
| 279 | + | repo, err := git.Open(e.dir, c.openOpts...) | |
| 182 | 280 | switch { | |
| 183 | 281 | case err == nil: | |
| 184 | 282 | op.repo = repo | |
| 283 | + | e.refreshed.Store(op.at.UnixNano()) // just opened: nothing to refresh yet | |
| 185 | 284 | case errors.Is(err, git.ErrUnsupported): | |
| 186 | 285 | c.onUnsupport(ctx, e.dir, name, err) | |
| 187 | 286 | default: | |
| @@ -204,10 +303,10 @@ func mapOpenErr(err error) error { | |||
| 204 | 303 | func (c *cache) markUnsupported(ctx context.Context, name domain.RepoName, e *entry, cause error) { | |
| 205 | 304 | c.onUnsupport(ctx, e.dir, name, cause) | |
| 206 | 305 | c.mu.Lock() | |
| 207 | - | defer c.mu.Unlock() | |
| 306 | + | defer c.unlock() | |
| 208 | 307 | if c.entries[e.dir] == e { | |
| 209 | 308 | c.dropLocked(e) | |
| 210 | - | n := &entry{dir: e.dir, ino: e.ino, dev: e.dev} | |
| 309 | + | n := &entry{dir: e.dir, ino: e.ino, dev: e.dev, lastUsed: time.Now()} | |
| 211 | 310 | n.once.Do(func() { n.opened.Store(&opened{at: time.Now()}) }) // no repo, no error: unsupported | |
| 212 | 311 | n.elem = c.lru.PushFront(n) | |
| 213 | 312 | c.entries[e.dir] = n | |
| @@ -222,14 +321,53 @@ func (h *handle) fallBack(ctx context.Context, name domain.RepoName, cause error | |||
| 222 | 321 | h.c.markUnsupported(ctx, name, h.e, cause) | |
| 223 | 322 | } | |
| 224 | 323 | ||
| 324 | + | // peek returns a handle on the open repository of dir, or nil when the cache | |
| 325 | + | // has none (it opens nothing and does not count as a use). | |
| 326 | + | func (c *cache) peek(dir string) *handle { | |
| 327 | + | c.mu.Lock() | |
| 328 | + | defer c.unlock() | |
| 329 | + | e := c.entries[dir] | |
| 330 | + | if c.closed || e == nil { | |
| 331 | + | return nil | |
| 332 | + | } | |
| 333 | + | if op := e.opened.Load(); op == nil || op.repo == nil { | |
| 334 | + | return nil | |
| 335 | + | } | |
| 336 | + | e.users++ | |
| 337 | + | return &handle{c: c, e: e, repo: e.opened.Load().repo} | |
| 338 | + | } | |
| 339 | + | ||
| 340 | + | // refreshNow runs Refresh on the open repository of dir, if the cache has it, | |
| 341 | + | // whatever the time of the last one. | |
| 342 | + | func (c *cache) refreshNow(ctx context.Context, name domain.RepoName, dir string) error { | |
| 343 | + | h := c.peek(dir) | |
| 344 | + | if h == nil { | |
| 345 | + | return nil | |
| 346 | + | } | |
| 347 | + | defer h.release() | |
| 348 | + | if err := h.repo.Refresh(); err != nil { | |
| 349 | + | if errors.Is(err, git.ErrUnsupported) { | |
| 350 | + | h.fallBack(ctx, name, err) | |
| 351 | + | return nil | |
| 352 | + | } | |
| 353 | + | return fmt.Errorf("refresh repository %q: %w", name, err) | |
| 354 | + | } | |
| 355 | + | h.e.refreshed.Store(time.Now().UnixNano()) | |
| 356 | + | return nil | |
| 357 | + | } | |
| 358 | + | ||
| 225 | 359 | func (c *cache) release(e *entry) { | |
| 226 | 360 | c.mu.Lock() | |
| 227 | 361 | e.users-- | |
| 228 | - | closeNow := e.dropped && e.users == 0 | |
| 229 | - | c.mu.Unlock() | |
| 230 | - | if closeNow { | |
| 231 | - | e.closeRepo() | |
| 362 | + | if e.dropped { | |
| 363 | + | if e.users == 0 { | |
| 364 | + | c.doomed = append(c.doomed, e) | |
| 365 | + | } | |
| 366 | + | } else { | |
| 367 | + | e.lastUsed = time.Now() // a long read is a use, up to its end | |
| 368 | + | c.lru.MoveToFront(e.elem) | |
| 232 | 369 | } | |
| 370 | + | c.unlock() | |
| 233 | 371 | } | |
| 234 | 372 | ||
| 235 | 373 | // closeRepo closes the repository of an entry that left the cache and has no | |
| @@ -241,8 +379,9 @@ func (e *entry) closeRepo() { | |||
| 241 | 379 | } | |
| 242 | 380 | } | |
| 243 | 381 | ||
| 244 | - | // dropLocked removes e from the cache. The caller holds the mutex. The | |
| 245 | - | // repository is closed by the last user (or by the caller if it has none). | |
| 382 | + | // dropLocked removes e from the cache. The caller holds the mutex and gives it | |
| 383 | + | // up with unlock, which closes the repository if e has no user; else the last | |
| 384 | + | // user closes it on release. | |
| 246 | 385 | func (c *cache) dropLocked(e *entry) { | |
| 247 | 386 | if e.dropped { | |
| 248 | 387 | return | |
| @@ -251,7 +390,7 @@ func (c *cache) dropLocked(e *entry) { | |||
| 251 | 390 | c.lru.Remove(e.elem) | |
| 252 | 391 | delete(c.entries, e.dir) | |
| 253 | 392 | if e.users == 0 { | |
| 254 | - | go e.closeRepo() // not under the mutex: Close takes the repo's own | |
| 393 | + | c.doomed = append(c.doomed, e) | |
| 255 | 394 | } | |
| 256 | 395 | } | |
| 257 | 396 | ||
| @@ -265,7 +404,7 @@ func (c *cache) evictLocked() { | |||
| 265 | 404 | // forgetEntry drops e if it is still the entry of its directory. | |
| 266 | 405 | func (c *cache) forgetEntry(e *entry) { | |
| 267 | 406 | c.mu.Lock() | |
| 268 | - | defer c.mu.Unlock() | |
| 407 | + | defer c.unlock() | |
| 269 | 408 | if c.entries[e.dir] == e { | |
| 270 | 409 | c.dropLocked(e) | |
| 271 | 410 | } | |
| @@ -274,16 +413,20 @@ func (c *cache) forgetEntry(e *entry) { | |||
| 274 | 413 | // forget drops the entry of dir. | |
| 275 | 414 | func (c *cache) forget(dir string) { | |
| 276 | 415 | c.mu.Lock() | |
| 277 | - | defer c.mu.Unlock() | |
| 416 | + | defer c.unlock() | |
| 278 | 417 | if e := c.entries[dir]; e != nil { | |
| 279 | 418 | c.dropLocked(e) | |
| 280 | 419 | } | |
| 281 | 420 | } | |
| 282 | 421 | ||
| 283 | - | // close drops every entry; later gets fail with ErrClosed. | |
| 422 | + | // close drops every entry; later gets fail with ErrClosed. The repositories | |
| 423 | + | // nobody uses are closed when it returns; the others when their last reader | |
| 424 | + | // is done. It ends the sweeper and waits for it. | |
| 284 | 425 | func (c *cache) close() { | |
| 426 | + | c.stopped.Do(func() { close(c.stop) }) | |
| 427 | + | c.swept.Wait() | |
| 285 | 428 | c.mu.Lock() | |
| 286 | - | defer c.mu.Unlock() | |
| 429 | + | defer c.unlock() | |
| 287 | 430 | c.closed = true | |
| 288 | 431 | for _, e := range c.entries { | |
| 289 | 432 | c.dropLocked(e) | |
| @@ -25,7 +25,22 @@ const DefaultMaxRepos = 64 | |||||
| 25 | 25 | ||||
| 26 | // DefaultRefreshEvery is how long a repository goes without a Refresh when | 26 | // DefaultRefreshEvery is how long a repository goes without a Refresh when | ||
| 27 | // Config.RefreshEvery is zero. See Config.RefreshEvery for what depends on it. | 27 | // Config.RefreshEvery is zero. See Config.RefreshEvery for what depends on it. | ||
| 28 | - | const DefaultRefreshEvery = time.Second | 28 | + | // A write that goes through ermite refreshes the repository at once (see |
| 29 | + | // Adapter.Refresh), so this is only the safety net for the other writers. | |||
| 30 | + | const DefaultRefreshEvery = 5 * time.Second | |||
| 31 | + | ||||
| 32 | + | // DefaultIdleTTL is how long a repository may go unused before it is closed, | |||
| 33 | + | // when Config.IdleTTL is zero. | |||
| 34 | + | const DefaultIdleTTL = 10 * time.Minute | |||
| 35 | + | ||||
| 36 | + | // DefaultDeltaCacheSize is the budget of the cache of delta bases of each open | |||
| 37 | + | // repository when Config.DeltaCacheSize is zero (the library's own is 32 MiB). | |||
| 38 | + | const DefaultDeltaCacheSize = 8 << 20 | |||
| 39 | + | ||||
| 40 | + | // DefaultMaxObjectSize is the largest object a repository holds in memory when | |||
| 41 | + | // Config.MaxObjectSize is zero (the library's own is 128 MiB). A bigger one | |||
| 42 | + | // is read by the wrapped reader. | |||
| 43 | + | const DefaultMaxObjectSize = 16 << 20 | |||
| 29 | 44 | ||||
| 30 | // unsupportedRetry is how long a repository that ermite-git cannot read is | 45 | // unsupportedRetry is how long a repository that ermite-git cannot read is | ||
| 31 | // served by git before Open is tried again (a repack, or the removal of a | 46 | // served by git before Open is tried again (a repack, or the removal of a | ||
| @@ -46,9 +61,10 @@ type entry struct { | |||||
| 46 | 61 | ||||
| 47 | refreshed atomic.Int64 // unix nanoseconds of the last Refresh | 62 | refreshed atomic.Int64 // unix nanoseconds of the last Refresh | ||
| 48 | 63 | ||||
| 49 | - | elem *list.Element | 64 | + | elem *list.Element |
| 50 | - | users int | 65 | + | users int |
| 51 | - | dropped bool // out of the cache: close when the last user is done | 66 | + | lastUsed time.Time // of the last get or release: the LRU is ordered by it |
| 67 | + | dropped bool // out of the cache: close when the last user is done | |||
| 52 | } | 68 | } | ||
| 53 | 69 | ||||
| 54 | // opened is what git.Open answered for an entry. | 70 | // opened is what git.Open answered for an entry. | ||
| @@ -69,25 +85,100 @@ type handle struct { | |||||
| 69 | func (h *handle) release() { h.c.release(h.e) } | 85 | func (h *handle) release() { h.c.release(h.e) } | ||
| 70 | 86 | ||||
| 71 | // cache is a bounded LRU of open repositories keyed by directory. A repository | 87 | // cache is a bounded LRU of open repositories keyed by directory. A repository | ||
| 72 | - | // leaves it by eviction, by Forget (the directory was deleted) or because the | 88 | + | // leaves it by eviction, by idling longer than the TTL, by Forget (the |
| 73 | - | // directory is no longer the same one (checked on each use through its | 89 | + | // directory was deleted) or because the directory is no longer the same one |
| 74 | - | // inode); it is closed once the reads that use it are done. | 90 | + | // (checked on each use through its inode); it is closed once the reads that |
| 91 | + | // use it are done, by the goroutine that made it leave or that was the last | |||
| 92 | + | // to use it. | |||
| 75 | type cache struct { | 93 | type cache struct { | ||
| 76 | max int | 94 | max int | ||
| 77 | refreshEvery time.Duration | 95 | refreshEvery time.Duration | ||
| 96 | + | idleTTL time.Duration // <= 0: never | |||
| 97 | + | openOpts []git.Option | |||
| 78 | onUnsupport func(ctx context.Context, dir string, name domain.RepoName, cause error) | 98 | onUnsupport func(ctx context.Context, dir string, name domain.RepoName, cause error) | ||
| 79 | 99 | ||||
| 80 | mu sync.Mutex | 100 | mu sync.Mutex | ||
| 81 | entries map[string]*entry | 101 | entries map[string]*entry | ||
| 82 | lru *list.List // front = most recently used | 102 | lru *list.List // front = most recently used | ||
| 103 | + | doomed []*entry // dropped without users: closed by unlock | |||
| 83 | closed bool | 104 | closed bool | ||
| 105 | + | ||||
| 106 | + | stop chan struct{} // closed by close; ends the sweeper | |||
| 107 | + | swept sync.WaitGroup | |||
| 108 | + | stopped sync.Once | |||
| 109 | + | } | |||
| 110 | + | ||||
| 111 | + | // cacheConfig is what newCache needs. | |||
| 112 | + | type cacheConfig struct { | |||
| 113 | + | max int | |||
| 114 | + | refreshEvery time.Duration | |||
| 115 | + | idleTTL time.Duration | |||
| 116 | + | openOpts []git.Option | |||
| 117 | + | onUnsupport func(context.Context, string, domain.RepoName, error) | |||
| 118 | + | } | |||
| 119 | + | ||||
| 120 | + | func newCache(cfg cacheConfig) *cache { | |||
| 121 | + | if cfg.max <= 0 { | |||
| 122 | + | cfg.max = DefaultMaxRepos | |||
| 123 | + | } | |||
| 124 | + | c := &cache{ | |||
| 125 | + | max: cfg.max, refreshEvery: cfg.refreshEvery, idleTTL: cfg.idleTTL, openOpts: cfg.openOpts, | |||
| 126 | + | onUnsupport: cfg.onUnsupport, entries: make(map[string]*entry), lru: list.New(), stop: make(chan struct{}), | |||
| 127 | + | } | |||
| 128 | + | if c.idleTTL > 0 { | |||
| 129 | + | c.swept.Add(1) | |||
| 130 | + | go c.sweeper() | |||
| 131 | + | } | |||
| 132 | + | return c | |||
| 133 | + | } | |||
| 134 | + | ||||
| 135 | + | // sweeper closes the repositories that idle past the TTL. It is the only | |||
| 136 | + | // goroutine of the cache, whatever the number of repositories, and close ends | |||
| 137 | + | // it. | |||
| 138 | + | func (c *cache) sweeper() { | |||
| 139 | + | defer c.swept.Done() | |||
| 140 | + | tick := time.NewTicker(max(c.idleTTL/4, time.Millisecond)) | |||
| 141 | + | defer tick.Stop() | |||
| 142 | + | for { | |||
| 143 | + | select { | |||
| 144 | + | case <-c.stop: | |||
| 145 | + | return | |||
| 146 | + | case now := <-tick.C: | |||
| 147 | + | c.sweep(now) | |||
| 148 | + | } | |||
| 149 | + | } | |||
| 150 | + | } | |||
| 151 | + | ||||
| 152 | + | // sweep drops the entries nobody used since now-idleTTL. The LRU is ordered by | |||
| 153 | + | // lastUsed, so the walk from the back stops at the first recent entry; one | |||
| 154 | + | // that is in use is not idle and stays. | |||
| 155 | + | func (c *cache) sweep(now time.Time) { | |||
| 156 | + | c.mu.Lock() | |||
| 157 | + | for el := c.lru.Back(); el != nil; { | |||
| 158 | + | prev := el.Prev() | |||
| 159 | + | e := el.Value.(*entry) | |||
| 160 | + | if now.Sub(e.lastUsed) <= c.idleTTL { | |||
| 161 | + | break | |||
| 162 | + | } | |||
| 163 | + | if e.users == 0 { | |||
| 164 | + | c.dropLocked(e) | |||
| 165 | + | } | |||
| 166 | + | el = prev | |||
| 167 | + | } | |||
| 168 | + | c.unlock() | |||
| 84 | } | 169 | } | ||
| 85 | 170 | ||||
| 86 | - | func newCache(max int, refreshEvery time.Duration, onUnsupport func(context.Context, string, domain.RepoName, error)) *cache { | 171 | + | // unlock releases the mutex, then closes what was dropped without users while |
| 87 | - | if max <= 0 { | 172 | + | // it was held: in the calling goroutine, so that when close returns the |
| 88 | - | max = DefaultMaxRepos | 173 | + | // mappings are released, and not under the mutex, since Close takes the |
| 174 | + | // repository's own. | |||
| 175 | + | func (c *cache) unlock() { | |||
| 176 | + | doomed := c.doomed | |||
| 177 | + | c.doomed = nil | |||
| 178 | + | c.mu.Unlock() | |||
| 179 | + | for _, e := range doomed { | |||
| 180 | + | e.closeRepo() | |||
| 89 | } | 181 | } | ||
| 90 | - | return &cache{max: max, refreshEvery: refreshEvery, onUnsupport: onUnsupport, entries: make(map[string]*entry), lru: list.New()} | |||
| 91 | } | 182 | } | ||
| 92 | 183 | ||||
| 93 | // get returns the entry of dir, opened and refreshed, or a nil repo when | 184 | // get returns the entry of dir, opened and refreshed, or a nil repo when | ||
| @@ -108,7 +199,7 @@ func (c *cache) get(ctx context.Context, name domain.RepoName, dir string) (*han | |||||
| 108 | 199 | ||||
| 109 | c.mu.Lock() | 200 | c.mu.Lock() | ||
| 110 | if c.closed { | 201 | if c.closed { | ||
| 111 | - | c.mu.Unlock() | 202 | + | c.unlock() |
| 112 | return nil, ErrClosed | 203 | return nil, ErrClosed | ||
| 113 | } | 204 | } | ||
| 114 | e := c.entries[dir] | 205 | e := c.entries[dir] | ||
| @@ -125,7 +216,8 @@ func (c *cache) get(ctx context.Context, name domain.RepoName, dir string) (*han | |||||
| 125 | c.lru.MoveToFront(e.elem) | 216 | c.lru.MoveToFront(e.elem) | ||
| 126 | } | 217 | } | ||
| 127 | e.users++ | 218 | e.users++ | ||
| 128 | - | c.mu.Unlock() | 219 | + | e.lastUsed = time.Now() |
| 220 | + | c.unlock() | |||
| 129 | 221 | ||||
| 130 | h := &handle{c: c, e: e} | 222 | h := &handle{c: c, e: e} | ||
| 131 | e.once.Do(func() { c.open(ctx, name, e) }) | 223 | e.once.Do(func() { c.open(ctx, name, e) }) | ||
| @@ -151,20 +243,26 @@ func (c *cache) get(ctx context.Context, name domain.RepoName, dir string) (*han | |||||
| 151 | } | 243 | } | ||
| 152 | 244 | ||||
| 153 | // refresh calls Refresh on the repository of e unless one ran less than | 245 | // refresh calls Refresh on the repository of e unless one ran less than | ||
| 154 | - | // refreshEvery ago (a negative interval refreshes on every read). The library | 246 | + | // refreshEvery ago (a negative interval refreshes on every read). Of the reads |
| 155 | - | // reads loose refs live, stat-checks packed-refs on each ref read and rescans | 247 | + | // that find it due, one wins a compare-and-swap on the time of the last |
| 156 | - | // the packs when an object is missing, so a read sees a push whether or not | 248 | + | // Refresh and runs it; the others go on with the snapshot they have. The |
| 157 | - | // Refresh ran; Refresh is what drops replaced packs, picks up a new | 249 | + | // library reads loose refs live, stat-checks packed-refs on each ref read and |
| 250 | + | // rescans the packs when an object is missing, so a read sees a push whether | |||
| 251 | + | // or not Refresh ran; Refresh is what drops replaced packs, picks up a new | |||
| 158 | // commit-graph and notices that the repository became unsupported. | 252 | // commit-graph and notices that the repository became unsupported. | ||
| 159 | func (c *cache) refresh(e *entry, repo *git.Repo) error { | 253 | func (c *cache) refresh(e *entry, repo *git.Repo) error { | ||
| 254 | + | if c.refreshEvery < 0 { | |||
| 255 | + | return repo.Refresh() | |||
| 256 | + | } | |||
| 160 | now := time.Now().UnixNano() | 257 | now := time.Now().UnixNano() | ||
| 161 | - | if c.refreshEvery >= 0 && now-e.refreshed.Load() < int64(c.refreshEvery) { | 258 | + | last := e.refreshed.Load() |
| 259 | + | if now-last < int64(c.refreshEvery) || !e.refreshed.CompareAndSwap(last, now) { | |||
| 162 | return nil | 260 | return nil | ||
| 163 | } | 261 | } | ||
| 164 | if err := repo.Refresh(); err != nil { | 262 | if err := repo.Refresh(); err != nil { | ||
| 263 | + | e.refreshed.CompareAndSwap(now, last) // the next read tries again | |||
| 165 | return err | 264 | return err | ||
| 166 | } | 265 | } | ||
| 167 | - | e.refreshed.Store(now) | |||
| 168 | return nil | 266 | return nil | ||
| 169 | } | 267 | } | ||
| 170 | 268 | ||||
| @@ -178,10 +276,11 @@ func (e *entry) stale() bool { | |||||
| 178 | // open runs git.Open once per entry; concurrent users wait in Once.Do. | 276 | // open runs git.Open once per entry; concurrent users wait in Once.Do. | ||
| 179 | func (c *cache) open(ctx context.Context, name domain.RepoName, e *entry) { | 277 | func (c *cache) open(ctx context.Context, name domain.RepoName, e *entry) { | ||
| 180 | op := &opened{at: time.Now()} | 278 | op := &opened{at: time.Now()} | ||
| 181 | - | repo, err := git.Open(e.dir) | 279 | + | repo, err := git.Open(e.dir, c.openOpts...) |
| 182 | switch { | 280 | switch { | ||
| 183 | case err == nil: | 281 | case err == nil: | ||
| 184 | op.repo = repo | 282 | op.repo = repo | ||
| 283 | + | e.refreshed.Store(op.at.UnixNano()) // just opened: nothing to refresh yet | |||
| 185 | case errors.Is(err, git.ErrUnsupported): | 284 | case errors.Is(err, git.ErrUnsupported): | ||
| 186 | c.onUnsupport(ctx, e.dir, name, err) | 285 | c.onUnsupport(ctx, e.dir, name, err) | ||
| 187 | default: | 286 | default: | ||
| @@ -204,10 +303,10 @@ func mapOpenErr(err error) error { | |||||
| 204 | func (c *cache) markUnsupported(ctx context.Context, name domain.RepoName, e *entry, cause error) { | 303 | func (c *cache) markUnsupported(ctx context.Context, name domain.RepoName, e *entry, cause error) { | ||
| 205 | c.onUnsupport(ctx, e.dir, name, cause) | 304 | c.onUnsupport(ctx, e.dir, name, cause) | ||
| 206 | c.mu.Lock() | 305 | c.mu.Lock() | ||
| 207 | - | defer c.mu.Unlock() | 306 | + | defer c.unlock() |
| 208 | if c.entries[e.dir] == e { | 307 | if c.entries[e.dir] == e { | ||
| 209 | c.dropLocked(e) | 308 | c.dropLocked(e) | ||
| 210 | - | n := &entry{dir: e.dir, ino: e.ino, dev: e.dev} | 309 | + | n := &entry{dir: e.dir, ino: e.ino, dev: e.dev, lastUsed: time.Now()} |
| 211 | n.once.Do(func() { n.opened.Store(&opened{at: time.Now()}) }) // no repo, no error: unsupported | 310 | n.once.Do(func() { n.opened.Store(&opened{at: time.Now()}) }) // no repo, no error: unsupported | ||
| 212 | n.elem = c.lru.PushFront(n) | 311 | n.elem = c.lru.PushFront(n) | ||
| 213 | c.entries[e.dir] = n | 312 | c.entries[e.dir] = n | ||
| @@ -222,14 +321,53 @@ func (h *handle) fallBack(ctx context.Context, name domain.RepoName, cause error | |||||
| 222 | h.c.markUnsupported(ctx, name, h.e, cause) | 321 | h.c.markUnsupported(ctx, name, h.e, cause) | ||
| 223 | } | 322 | } | ||
| 224 | 323 | ||||
| 324 | + | // peek returns a handle on the open repository of dir, or nil when the cache | |||
| 325 | + | // has none (it opens nothing and does not count as a use). | |||
| 326 | + | func (c *cache) peek(dir string) *handle { | |||
| 327 | + | c.mu.Lock() | |||
| 328 | + | defer c.unlock() | |||
| 329 | + | e := c.entries[dir] | |||
| 330 | + | if c.closed || e == nil { | |||
| 331 | + | return nil | |||
| 332 | + | } | |||
| 333 | + | if op := e.opened.Load(); op == nil || op.repo == nil { | |||
| 334 | + | return nil | |||
| 335 | + | } | |||
| 336 | + | e.users++ | |||
| 337 | + | return &handle{c: c, e: e, repo: e.opened.Load().repo} | |||
| 338 | + | } | |||
| 339 | + | ||||
| 340 | + | // refreshNow runs Refresh on the open repository of dir, if the cache has it, | |||
| 341 | + | // whatever the time of the last one. | |||
| 342 | + | func (c *cache) refreshNow(ctx context.Context, name domain.RepoName, dir string) error { | |||
| 343 | + | h := c.peek(dir) | |||
| 344 | + | if h == nil { | |||
| 345 | + | return nil | |||
| 346 | + | } | |||
| 347 | + | defer h.release() | |||
| 348 | + | if err := h.repo.Refresh(); err != nil { | |||
| 349 | + | if errors.Is(err, git.ErrUnsupported) { | |||
| 350 | + | h.fallBack(ctx, name, err) | |||
| 351 | + | return nil | |||
| 352 | + | } | |||
| 353 | + | return fmt.Errorf("refresh repository %q: %w", name, err) | |||
| 354 | + | } | |||
| 355 | + | h.e.refreshed.Store(time.Now().UnixNano()) | |||
| 356 | + | return nil | |||
| 357 | + | } | |||
| 358 | + | ||||
| 225 | func (c *cache) release(e *entry) { | 359 | func (c *cache) release(e *entry) { | ||
| 226 | c.mu.Lock() | 360 | c.mu.Lock() | ||
| 227 | e.users-- | 361 | e.users-- | ||
| 228 | - | closeNow := e.dropped && e.users == 0 | 362 | + | if e.dropped { |
| 229 | - | c.mu.Unlock() | 363 | + | if e.users == 0 { |
| 230 | - | if closeNow { | 364 | + | c.doomed = append(c.doomed, e) |
| 231 | - | e.closeRepo() | 365 | + | } |
| 366 | + | } else { | |||
| 367 | + | e.lastUsed = time.Now() // a long read is a use, up to its end | |||
| 368 | + | c.lru.MoveToFront(e.elem) | |||
| 232 | } | 369 | } | ||
| 370 | + | c.unlock() | |||
| 233 | } | 371 | } | ||
| 234 | 372 | ||||
| 235 | // closeRepo closes the repository of an entry that left the cache and has no | 373 | // closeRepo closes the repository of an entry that left the cache and has no | ||
| @@ -241,8 +379,9 @@ func (e *entry) closeRepo() { | |||||
| 241 | } | 379 | } | ||
| 242 | } | 380 | } | ||
| 243 | 381 | ||||
| 244 | - | // dropLocked removes e from the cache. The caller holds the mutex. The | 382 | + | // dropLocked removes e from the cache. The caller holds the mutex and gives it |
| 245 | - | // repository is closed by the last user (or by the caller if it has none). | 383 | + | // up with unlock, which closes the repository if e has no user; else the last |
| 384 | + | // user closes it on release. | |||
| 246 | func (c *cache) dropLocked(e *entry) { | 385 | func (c *cache) dropLocked(e *entry) { | ||
| 247 | if e.dropped { | 386 | if e.dropped { | ||
| 248 | return | 387 | return | ||
| @@ -251,7 +390,7 @@ func (c *cache) dropLocked(e *entry) { | |||||
| 251 | c.lru.Remove(e.elem) | 390 | c.lru.Remove(e.elem) | ||
| 252 | delete(c.entries, e.dir) | 391 | delete(c.entries, e.dir) | ||
| 253 | if e.users == 0 { | 392 | if e.users == 0 { | ||
| 254 | - | go e.closeRepo() // not under the mutex: Close takes the repo's own | 393 | + | c.doomed = append(c.doomed, e) |
| 255 | } | 394 | } | ||
| 256 | } | 395 | } | ||
| 257 | 396 | ||||
| @@ -265,7 +404,7 @@ func (c *cache) evictLocked() { | |||||
| 265 | // forgetEntry drops e if it is still the entry of its directory. | 404 | // forgetEntry drops e if it is still the entry of its directory. | ||
| 266 | func (c *cache) forgetEntry(e *entry) { | 405 | func (c *cache) forgetEntry(e *entry) { | ||
| 267 | c.mu.Lock() | 406 | c.mu.Lock() | ||
| 268 | - | defer c.mu.Unlock() | 407 | + | defer c.unlock() |
| 269 | if c.entries[e.dir] == e { | 408 | if c.entries[e.dir] == e { | ||
| 270 | c.dropLocked(e) | 409 | c.dropLocked(e) | ||
| 271 | } | 410 | } | ||
| @@ -274,16 +413,20 @@ func (c *cache) forgetEntry(e *entry) { | |||||
| 274 | // forget drops the entry of dir. | 413 | // forget drops the entry of dir. | ||
| 275 | func (c *cache) forget(dir string) { | 414 | func (c *cache) forget(dir string) { | ||
| 276 | c.mu.Lock() | 415 | c.mu.Lock() | ||
| 277 | - | defer c.mu.Unlock() | 416 | + | defer c.unlock() |
| 278 | if e := c.entries[dir]; e != nil { | 417 | if e := c.entries[dir]; e != nil { | ||
| 279 | c.dropLocked(e) | 418 | c.dropLocked(e) | ||
| 280 | } | 419 | } | ||
| 281 | } | 420 | } | ||
| 282 | 421 | ||||
| 283 | - | // close drops every entry; later gets fail with ErrClosed. | 422 | + | // close drops every entry; later gets fail with ErrClosed. The repositories |
| 423 | + | // nobody uses are closed when it returns; the others when their last reader | |||
| 424 | + | // is done. It ends the sweeper and waits for it. | |||
| 284 | func (c *cache) close() { | 425 | func (c *cache) close() { | ||
| 426 | + | c.stopped.Do(func() { close(c.stop) }) | |||
| 427 | + | c.swept.Wait() | |||
| 285 | c.mu.Lock() | 428 | c.mu.Lock() | ||
| 286 | - | defer c.mu.Unlock() | 429 | + | defer c.unlock() |
| 287 | c.closed = true | 430 | c.closed = true | ||
| 288 | for _, e := range c.entries { | 431 | for _, e := range c.entries { | ||
| 289 | c.dropLocked(e) | 432 | c.dropLocked(e) | ||
M internal/adapters/ermitegit/cache_test.go +102 -0
| @@ -6,10 +6,12 @@ import ( | |||
| 6 | 6 | "bytes" | |
| 7 | 7 | "context" | |
| 8 | 8 | "fmt" | |
| 9 | + | "io" | |
| 9 | 10 | "os" | |
| 10 | 11 | "path/filepath" | |
| 11 | 12 | "sync" | |
| 12 | 13 | "testing" | |
| 14 | + | "time" | |
| 13 | 15 | ||
| 14 | 16 | "github.com/bnema/zerowrap" | |
| 15 | 17 | "github.com/stretchr/testify/assert" | |
| @@ -204,3 +206,103 @@ func TestEncodedCommitIsServedByGit(t *testing.T) { | |||
| 204 | 206 | assert.Equal(t, want, got) | |
| 205 | 207 | assert.Equal(t, "café", got.Commits[0].Message) | |
| 206 | 208 | } | |
| 209 | + | ||
| 210 | + | // A repository that nobody reads for IdleTTL is closed, without a goroutine | |
| 211 | + | // per repository. | |
| 212 | + | func TestIdleRepositoriesAreClosed(t *testing.T) { | |
| 213 | + | ctx := context.Background() | |
| 214 | + | a, _, data := newAdapter(t, ermitegit.AllRoutes(ermitegit.ErmiteGit), func(c *ermitegit.Config) { c.IdleTTL = 80 * time.Millisecond }) | |
| 215 | + | f := commitOf(t, data, "idle") | |
| 216 | + | _, err := a.ListRefs(ctx, f.Name) | |
| 217 | + | require.NoError(t, err) | |
| 218 | + | require.Equal(t, 1, a.OpenRepos()) | |
| 219 | + | require.Eventually(t, func() bool { return a.OpenRepos() == 0 }, 5*time.Second, 10*time.Millisecond) | |
| 220 | + | ||
| 221 | + | // It opens again on the next read. | |
| 222 | + | _, err = a.ListRefs(ctx, f.Name) | |
| 223 | + | require.NoError(t, err) | |
| 224 | + | assert.Equal(t, 1, a.OpenRepos()) | |
| 225 | + | } | |
| 226 | + | ||
| 227 | + | // The idle close does not cut a blob stream the reader still holds: the library | |
| 228 | + | // keeps the stream valid after Close. | |
| 229 | + | func TestIdleCloseDoesNotCutAStream(t *testing.T) { | |
| 230 | + | ctx := context.Background() | |
| 231 | + | a, _, data := newAdapter(t, ermitegit.AllRoutes(ermitegit.ErmiteGit), func(c *ermitegit.Config) { c.IdleTTL = 50 * time.Millisecond }) | |
| 232 | + | f := readertest.NewFixture(t, data, "stream") | |
| 233 | + | content := readertest.RandomBytes(3<<20, 12) | |
| 234 | + | f.WriteBytes("big.bin", content) | |
| 235 | + | f.Commit("big") | |
| 236 | + | ||
| 237 | + | _, rc, err := a.OpenBlob(ctx, f.Name, domain.Revision{}, readertest.TreePath(t, "big.bin")) | |
| 238 | + | require.NoError(t, err) | |
| 239 | + | require.Eventually(t, func() bool { return a.OpenRepos() == 0 }, 5*time.Second, 10*time.Millisecond) | |
| 240 | + | got, err := io.ReadAll(rc) | |
| 241 | + | require.NoError(t, err) | |
| 242 | + | require.NoError(t, rc.Close()) | |
| 243 | + | assert.True(t, bytes.Equal(content, got)) | |
| 244 | + | } | |
| 245 | + | ||
| 246 | + | // mapped reports whether the process maps a file under dir (Linux). | |
| 247 | + | func mapped(t *testing.T, dir string) bool { | |
| 248 | + | t.Helper() | |
| 249 | + | maps, err := os.ReadFile("/proc/self/maps") | |
| 250 | + | if err != nil { | |
| 251 | + | t.Skip("no /proc/self/maps") | |
| 252 | + | } | |
| 253 | + | return bytes.Contains(maps, []byte(dir)) | |
| 254 | + | } | |
| 255 | + | ||
| 256 | + | // Close releases the mappings before it returns, not later from another | |
| 257 | + | // goroutine. | |
| 258 | + | func TestCloseReleasesMappingsSynchronously(t *testing.T) { | |
| 259 | + | ctx := context.Background() | |
| 260 | + | a, _, data := newAdapter(t, ermitegit.AllRoutes(ermitegit.ErmiteGit), nil) | |
| 261 | + | h := readertest.NewHistory(t, data) | |
| 262 | + | h.Maintain() | |
| 263 | + | _, err := a.ListCommits(ctx, h.Name, ports.CommitQuery{Limit: 5}) | |
| 264 | + | require.NoError(t, err) | |
| 265 | + | require.True(t, mapped(t, h.Bare), "the commit-graph is mapped while the repository is open") | |
| 266 | + | a.Close() | |
| 267 | + | assert.False(t, mapped(t, h.Bare), "Close returned with the mappings still there") | |
| 268 | + | } | |
| 269 | + | ||
| 270 | + | // Forget (a deleted repository) also closes before it returns. | |
| 271 | + | func TestForgetReleasesMappingsSynchronously(t *testing.T) { | |
| 272 | + | ctx := context.Background() | |
| 273 | + | a, _, data := newAdapter(t, ermitegit.AllRoutes(ermitegit.ErmiteGit), nil) | |
| 274 | + | h := readertest.NewHistory(t, data) | |
| 275 | + | h.Maintain() | |
| 276 | + | _, err := a.ListCommits(ctx, h.Name, ports.CommitQuery{Limit: 5}) | |
| 277 | + | require.NoError(t, err) | |
| 278 | + | require.True(t, mapped(t, h.Bare)) | |
| 279 | + | a.Forget(h.Bare) | |
| 280 | + | assert.False(t, mapped(t, h.Bare)) | |
| 281 | + | } | |
| 282 | + | ||
| 283 | + | // Refresh, called after a write of ermite, makes the library notice at once | |
| 284 | + | // what the throttled read path would not for an hour. | |
| 285 | + | func TestRefreshAfterAWrite(t *testing.T) { | |
| 286 | + | var logs bytes.Buffer | |
| 287 | + | log := zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &logs}) | |
| 288 | + | ctx := zerowrap.WithCtx(context.Background(), log) | |
| 289 | + | a, g, data := newAdapter(t, ermitegit.AllRoutes(ermitegit.ErmiteGit), func(c *ermitegit.Config) { c.RefreshEvery = time.Hour }) | |
| 290 | + | h := readertest.NewHistory(t, data) | |
| 291 | + | ||
| 292 | + | a.Refresh(ctx, h.Name) // not open: nothing to do | |
| 293 | + | assert.Equal(t, 0, a.OpenRepos()) | |
| 294 | + | _, err := a.ListRefs(ctx, h.Name) | |
| 295 | + | require.NoError(t, err) | |
| 296 | + | ||
| 297 | + | h.Git("replace", h.C2, h.C3) // the repository stops being readable by the library | |
| 298 | + | a.Refresh(ctx, h.Name) | |
| 299 | + | assert.Contains(t, logs.String(), "served by git", "Refresh noticed the replace ref") | |
| 300 | + | want, err := g.ListRefs(ctx, h.Name) | |
| 301 | + | require.NoError(t, err) | |
| 302 | + | got, err := a.ListRefs(ctx, h.Name) | |
| 303 | + | require.NoError(t, err) | |
| 304 | + | assert.Equal(t, want, got) | |
| 305 | + | ||
| 306 | + | a.Close() | |
| 307 | + | a.Refresh(ctx, h.Name) // after Close: no-op | |
| 308 | + | } | |
| @@ -6,10 +6,12 @@ import ( | |||||
| 6 | "bytes" | 6 | "bytes" | ||
| 7 | "context" | 7 | "context" | ||
| 8 | "fmt" | 8 | "fmt" | ||
| 9 | + | "io" | |||
| 9 | "os" | 10 | "os" | ||
| 10 | "path/filepath" | 11 | "path/filepath" | ||
| 11 | "sync" | 12 | "sync" | ||
| 12 | "testing" | 13 | "testing" | ||
| 14 | + | "time" | |||
| 13 | 15 | ||||
| 14 | "github.com/bnema/zerowrap" | 16 | "github.com/bnema/zerowrap" | ||
| 15 | "github.com/stretchr/testify/assert" | 17 | "github.com/stretchr/testify/assert" | ||
| @@ -204,3 +206,103 @@ func TestEncodedCommitIsServedByGit(t *testing.T) { | |||||
| 204 | assert.Equal(t, want, got) | 206 | assert.Equal(t, want, got) | ||
| 205 | assert.Equal(t, "café", got.Commits[0].Message) | 207 | assert.Equal(t, "café", got.Commits[0].Message) | ||
| 206 | } | 208 | } | ||
| 209 | + | ||||
| 210 | + | // A repository that nobody reads for IdleTTL is closed, without a goroutine | |||
| 211 | + | // per repository. | |||
| 212 | + | func TestIdleRepositoriesAreClosed(t *testing.T) { | |||
| 213 | + | ctx := context.Background() | |||
| 214 | + | a, _, data := newAdapter(t, ermitegit.AllRoutes(ermitegit.ErmiteGit), func(c *ermitegit.Config) { c.IdleTTL = 80 * time.Millisecond }) | |||
| 215 | + | f := commitOf(t, data, "idle") | |||
| 216 | + | _, err := a.ListRefs(ctx, f.Name) | |||
| 217 | + | require.NoError(t, err) | |||
| 218 | + | require.Equal(t, 1, a.OpenRepos()) | |||
| 219 | + | require.Eventually(t, func() bool { return a.OpenRepos() == 0 }, 5*time.Second, 10*time.Millisecond) | |||
| 220 | + | ||||
| 221 | + | // It opens again on the next read. | |||
| 222 | + | _, err = a.ListRefs(ctx, f.Name) | |||
| 223 | + | require.NoError(t, err) | |||
| 224 | + | assert.Equal(t, 1, a.OpenRepos()) | |||
| 225 | + | } | |||
| 226 | + | ||||
| 227 | + | // The idle close does not cut a blob stream the reader still holds: the library | |||
| 228 | + | // keeps the stream valid after Close. | |||
| 229 | + | func TestIdleCloseDoesNotCutAStream(t *testing.T) { | |||
| 230 | + | ctx := context.Background() | |||
| 231 | + | a, _, data := newAdapter(t, ermitegit.AllRoutes(ermitegit.ErmiteGit), func(c *ermitegit.Config) { c.IdleTTL = 50 * time.Millisecond }) | |||
| 232 | + | f := readertest.NewFixture(t, data, "stream") | |||
| 233 | + | content := readertest.RandomBytes(3<<20, 12) | |||
| 234 | + | f.WriteBytes("big.bin", content) | |||
| 235 | + | f.Commit("big") | |||
| 236 | + | ||||
| 237 | + | _, rc, err := a.OpenBlob(ctx, f.Name, domain.Revision{}, readertest.TreePath(t, "big.bin")) | |||
| 238 | + | require.NoError(t, err) | |||
| 239 | + | require.Eventually(t, func() bool { return a.OpenRepos() == 0 }, 5*time.Second, 10*time.Millisecond) | |||
| 240 | + | got, err := io.ReadAll(rc) | |||
| 241 | + | require.NoError(t, err) | |||
| 242 | + | require.NoError(t, rc.Close()) | |||
| 243 | + | assert.True(t, bytes.Equal(content, got)) | |||
| 244 | + | } | |||
| 245 | + | ||||
| 246 | + | // mapped reports whether the process maps a file under dir (Linux). | |||
| 247 | + | func mapped(t *testing.T, dir string) bool { | |||
| 248 | + | t.Helper() | |||
| 249 | + | maps, err := os.ReadFile("/proc/self/maps") | |||
| 250 | + | if err != nil { | |||
| 251 | + | t.Skip("no /proc/self/maps") | |||
| 252 | + | } | |||
| 253 | + | return bytes.Contains(maps, []byte(dir)) | |||
| 254 | + | } | |||
| 255 | + | ||||
| 256 | + | // Close releases the mappings before it returns, not later from another | |||
| 257 | + | // goroutine. | |||
| 258 | + | func TestCloseReleasesMappingsSynchronously(t *testing.T) { | |||
| 259 | + | ctx := context.Background() | |||
| 260 | + | a, _, data := newAdapter(t, ermitegit.AllRoutes(ermitegit.ErmiteGit), nil) | |||
| 261 | + | h := readertest.NewHistory(t, data) | |||
| 262 | + | h.Maintain() | |||
| 263 | + | _, err := a.ListCommits(ctx, h.Name, ports.CommitQuery{Limit: 5}) | |||
| 264 | + | require.NoError(t, err) | |||
| 265 | + | require.True(t, mapped(t, h.Bare), "the commit-graph is mapped while the repository is open") | |||
| 266 | + | a.Close() | |||
| 267 | + | assert.False(t, mapped(t, h.Bare), "Close returned with the mappings still there") | |||
| 268 | + | } | |||
| 269 | + | ||||
| 270 | + | // Forget (a deleted repository) also closes before it returns. | |||
| 271 | + | func TestForgetReleasesMappingsSynchronously(t *testing.T) { | |||
| 272 | + | ctx := context.Background() | |||
| 273 | + | a, _, data := newAdapter(t, ermitegit.AllRoutes(ermitegit.ErmiteGit), nil) | |||
| 274 | + | h := readertest.NewHistory(t, data) | |||
| 275 | + | h.Maintain() | |||
| 276 | + | _, err := a.ListCommits(ctx, h.Name, ports.CommitQuery{Limit: 5}) | |||
| 277 | + | require.NoError(t, err) | |||
| 278 | + | require.True(t, mapped(t, h.Bare)) | |||
| 279 | + | a.Forget(h.Bare) | |||
| 280 | + | assert.False(t, mapped(t, h.Bare)) | |||
| 281 | + | } | |||
| 282 | + | ||||
| 283 | + | // Refresh, called after a write of ermite, makes the library notice at once | |||
| 284 | + | // what the throttled read path would not for an hour. | |||
| 285 | + | func TestRefreshAfterAWrite(t *testing.T) { | |||
| 286 | + | var logs bytes.Buffer | |||
| 287 | + | log := zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &logs}) | |||
| 288 | + | ctx := zerowrap.WithCtx(context.Background(), log) | |||
| 289 | + | a, g, data := newAdapter(t, ermitegit.AllRoutes(ermitegit.ErmiteGit), func(c *ermitegit.Config) { c.RefreshEvery = time.Hour }) | |||
| 290 | + | h := readertest.NewHistory(t, data) | |||
| 291 | + | ||||
| 292 | + | a.Refresh(ctx, h.Name) // not open: nothing to do | |||
| 293 | + | assert.Equal(t, 0, a.OpenRepos()) | |||
| 294 | + | _, err := a.ListRefs(ctx, h.Name) | |||
| 295 | + | require.NoError(t, err) | |||
| 296 | + | ||||
| 297 | + | h.Git("replace", h.C2, h.C3) // the repository stops being readable by the library | |||
| 298 | + | a.Refresh(ctx, h.Name) | |||
| 299 | + | assert.Contains(t, logs.String(), "served by git", "Refresh noticed the replace ref") | |||
| 300 | + | want, err := g.ListRefs(ctx, h.Name) | |||
| 301 | + | require.NoError(t, err) | |||
| 302 | + | got, err := a.ListRefs(ctx, h.Name) | |||
| 303 | + | require.NoError(t, err) | |||
| 304 | + | assert.Equal(t, want, got) | |||
| 305 | + | ||||
| 306 | + | a.Close() | |||
| 307 | + | a.Refresh(ctx, h.Name) // after Close: no-op | |||
| 308 | + | } | |||
M internal/adapters/ermitegit/ermitegit.go +57 -8
| @@ -19,12 +19,19 @@ | |||
| 19 | 19 | // whichever path made it, because the library reads refs from disk on each | |
| 20 | 20 | // call (loose refs always, packed-refs after a stat) and rescans the packs | |
| 21 | 21 | // when an object is missing (TestFreshness proves it with no Refresh at all). | |
| 22 | - | // The adapter still calls Refresh, at most once per Config.RefreshEvery for | |
| 23 | - | // each repository, to release replaced packs, pick up a new commit-graph and | |
| 24 | - | // notice a repository that stopped being supported; a stat-based Refresh is | |
| 25 | - | // 20 µs and 34 allocations, too much to pay on every read of a method that | |
| 26 | - | // itself costs 20 µs. The cache of open repositories is bounded (LRU); | |
| 27 | - | // gitcli's Trash tells the adapter to drop a deleted repository (see Forget). | |
| 22 | + | // The adapter still calls Refresh to release replaced packs, pick up a new | |
| 23 | + | // commit-graph and notice a repository that stopped being supported: at once | |
| 24 | + | // when ermite itself wrote (the push and the maintenance tell it through | |
| 25 | + | // Refresh), and as a safety net at most once per Config.RefreshEvery for each | |
| 26 | + | // repository while it is read, by one reader at a time. A stat-based Refresh | |
| 27 | + | // is 20 µs and 34 allocations, too much to pay on every read of a method that | |
| 28 | + | // itself costs 20 µs. | |
| 29 | + | // | |
| 30 | + | // Memory: the cache of open repositories is bounded (LRU) and drops the ones | |
| 31 | + | // idle longer than Config.IdleTTL; each repository holds at most | |
| 32 | + | // Config.DeltaCacheSize of delta bases and reads objects up to | |
| 33 | + | // Config.MaxObjectSize into memory. gitcli's Trash tells the adapter to drop | |
| 34 | + | // a deleted repository (see Forget). | |
| 28 | 35 | package ermitegit | |
| 29 | 36 | ||
| 30 | 37 | import ( | |
| @@ -42,7 +49,10 @@ import ( | |||
| 42 | 49 | "bnema.dev/ermite/internal/ports" | |
| 43 | 50 | ) | |
| 44 | 51 | ||
| 45 | - | var _ ports.RepoReader = (*Adapter)(nil) | |
| 52 | + | var ( | |
| 53 | + | _ ports.RepoReader = (*Adapter)(nil) | |
| 54 | + | _ ports.RepoFreshener = (*Adapter)(nil) | |
| 55 | + | ) | |
| 46 | 56 | ||
| 47 | 57 | // ErrClosed is returned by reads after Close. | |
| 48 | 58 | var ErrClosed = errors.New("ermitegit: reader is closed") | |
| @@ -107,6 +117,18 @@ type Config struct { | |||
| 107 | 117 | RefreshEvery time.Duration | |
| 108 | 118 | // MaxRepos bounds the repositories kept open; zero means DefaultMaxRepos. | |
| 109 | 119 | MaxRepos int | |
| 120 | + | // IdleTTL is how long a repository may go unused before it is closed. | |
| 121 | + | // Zero means DefaultIdleTTL; a negative value keeps repositories open | |
| 122 | + | // until evicted. | |
| 123 | + | IdleTTL time.Duration | |
| 124 | + | // DeltaCacheSize is the budget in bytes of the cache of delta bases of | |
| 125 | + | // each open repository. Zero means DefaultDeltaCacheSize; a negative value | |
| 126 | + | // disables the cache. | |
| 127 | + | DeltaCacheSize int64 | |
| 128 | + | // MaxObjectSize is the size in bytes above which an object that must be | |
| 129 | + | // held in memory is read by the wrapped reader instead. Zero means | |
| 130 | + | // DefaultMaxObjectSize. | |
| 131 | + | MaxObjectSize int64 | |
| 110 | 132 | // Routes overrides DefaultRoutes when not nil. | |
| 111 | 133 | Routes *Routes | |
| 112 | 134 | } | |
| @@ -132,8 +154,24 @@ func New(cfg Config) *Adapter { | |||
| 132 | 154 | if refreshEvery == 0 { | |
| 133 | 155 | refreshEvery = DefaultRefreshEvery | |
| 134 | 156 | } | |
| 157 | + | idleTTL := cfg.IdleTTL | |
| 158 | + | if idleTTL == 0 { | |
| 159 | + | idleTTL = DefaultIdleTTL | |
| 160 | + | } | |
| 161 | + | deltaCache := cfg.DeltaCacheSize | |
| 162 | + | if deltaCache == 0 { | |
| 163 | + | deltaCache = DefaultDeltaCacheSize | |
| 164 | + | } | |
| 165 | + | maxObject := cfg.MaxObjectSize | |
| 166 | + | if maxObject == 0 { | |
| 167 | + | maxObject = DefaultMaxObjectSize | |
| 168 | + | } | |
| 135 | 169 | a := &Adapter{cli: cfg.Fallback, dirOf: cfg.RepoDir, routes: routes} | |
| 136 | - | a.cache = newCache(cfg.MaxRepos, refreshEvery, a.logUnsupported) | |
| 170 | + | a.cache = newCache(cacheConfig{ | |
| 171 | + | max: cfg.MaxRepos, refreshEvery: refreshEvery, idleTTL: idleTTL, | |
| 172 | + | openOpts: []git.Option{git.WithDeltaCacheSize(deltaCache), git.WithMaxObjectSize(maxObject)}, | |
| 173 | + | onUnsupport: a.logUnsupported, | |
| 174 | + | }) | |
| 137 | 175 | return a | |
| 138 | 176 | } | |
| 139 | 177 | ||
| @@ -148,6 +186,17 @@ func (a *Adapter) Forget(dir string) { | |||
| 148 | 186 | a.logged.Delete(dir) | |
| 149 | 187 | } | |
| 150 | 188 | ||
| 189 | + | // Refresh tells the adapter that ermite wrote to the repository (a push, the | |
| 190 | + | // maintenance): the library refreshes it now, if it is open, instead of at the | |
| 191 | + | // next read after RefreshEvery. A repository that is not open has nothing to | |
| 192 | + | // refresh. A failure is logged, not returned: reads do not depend on it. | |
| 193 | + | func (a *Adapter) Refresh(ctx context.Context, repo domain.RepoName) { | |
| 194 | + | if err := a.cache.refreshNow(ctx, repo, a.dirOf(repo)); err != nil { | |
| 195 | + | log := zerowrap.FromCtx(ctx) | |
| 196 | + | log.Warn().Err(err).Str("repo", repo.String()).Msg("refresh of the open repository failed") | |
| 197 | + | } | |
| 198 | + | } | |
| 199 | + | ||
| 151 | 200 | // OpenRepos returns the number of repositories in the cache, for tests. | |
| 152 | 201 | func (a *Adapter) OpenRepos() int { return a.cache.size() } | |
| 153 | 202 | ||
| @@ -19,12 +19,19 @@ | |||||
| 19 | // whichever path made it, because the library reads refs from disk on each | 19 | // whichever path made it, because the library reads refs from disk on each | ||
| 20 | // call (loose refs always, packed-refs after a stat) and rescans the packs | 20 | // call (loose refs always, packed-refs after a stat) and rescans the packs | ||
| 21 | // when an object is missing (TestFreshness proves it with no Refresh at all). | 21 | // when an object is missing (TestFreshness proves it with no Refresh at all). | ||
| 22 | - | // The adapter still calls Refresh, at most once per Config.RefreshEvery for | 22 | + | // The adapter still calls Refresh to release replaced packs, pick up a new |
| 23 | - | // each repository, to release replaced packs, pick up a new commit-graph and | 23 | + | // commit-graph and notice a repository that stopped being supported: at once |
| 24 | - | // notice a repository that stopped being supported; a stat-based Refresh is | 24 | + | // when ermite itself wrote (the push and the maintenance tell it through |
| 25 | - | // 20 µs and 34 allocations, too much to pay on every read of a method that | 25 | + | // Refresh), and as a safety net at most once per Config.RefreshEvery for each |
| 26 | - | // itself costs 20 µs. The cache of open repositories is bounded (LRU); | 26 | + | // repository while it is read, by one reader at a time. A stat-based Refresh |
| 27 | - | // gitcli's Trash tells the adapter to drop a deleted repository (see Forget). | 27 | + | // is 20 µs and 34 allocations, too much to pay on every read of a method that |
| 28 | + | // itself costs 20 µs. | |||
| 29 | + | // | |||
| 30 | + | // Memory: the cache of open repositories is bounded (LRU) and drops the ones | |||
| 31 | + | // idle longer than Config.IdleTTL; each repository holds at most | |||
| 32 | + | // Config.DeltaCacheSize of delta bases and reads objects up to | |||
| 33 | + | // Config.MaxObjectSize into memory. gitcli's Trash tells the adapter to drop | |||
| 34 | + | // a deleted repository (see Forget). | |||
| 28 | package ermitegit | 35 | package ermitegit | ||
| 29 | 36 | ||||
| 30 | import ( | 37 | import ( | ||
| @@ -42,7 +49,10 @@ import ( | |||||
| 42 | "bnema.dev/ermite/internal/ports" | 49 | "bnema.dev/ermite/internal/ports" | ||
| 43 | ) | 50 | ) | ||
| 44 | 51 | ||||
| 45 | - | var _ ports.RepoReader = (*Adapter)(nil) | 52 | + | var ( |
| 53 | + | _ ports.RepoReader = (*Adapter)(nil) | |||
| 54 | + | _ ports.RepoFreshener = (*Adapter)(nil) | |||
| 55 | + | ) | |||
| 46 | 56 | ||||
| 47 | // ErrClosed is returned by reads after Close. | 57 | // ErrClosed is returned by reads after Close. | ||
| 48 | var ErrClosed = errors.New("ermitegit: reader is closed") | 58 | var ErrClosed = errors.New("ermitegit: reader is closed") | ||
| @@ -107,6 +117,18 @@ type Config struct { | |||||
| 107 | RefreshEvery time.Duration | 117 | RefreshEvery time.Duration | ||
| 108 | // MaxRepos bounds the repositories kept open; zero means DefaultMaxRepos. | 118 | // MaxRepos bounds the repositories kept open; zero means DefaultMaxRepos. | ||
| 109 | MaxRepos int | 119 | MaxRepos int | ||
| 120 | + | // IdleTTL is how long a repository may go unused before it is closed. | |||
| 121 | + | // Zero means DefaultIdleTTL; a negative value keeps repositories open | |||
| 122 | + | // until evicted. | |||
| 123 | + | IdleTTL time.Duration | |||
| 124 | + | // DeltaCacheSize is the budget in bytes of the cache of delta bases of | |||
| 125 | + | // each open repository. Zero means DefaultDeltaCacheSize; a negative value | |||
| 126 | + | // disables the cache. | |||
| 127 | + | DeltaCacheSize int64 | |||
| 128 | + | // MaxObjectSize is the size in bytes above which an object that must be | |||
| 129 | + | // held in memory is read by the wrapped reader instead. Zero means | |||
| 130 | + | // DefaultMaxObjectSize. | |||
| 131 | + | MaxObjectSize int64 | |||
| 110 | // Routes overrides DefaultRoutes when not nil. | 132 | // Routes overrides DefaultRoutes when not nil. | ||
| 111 | Routes *Routes | 133 | Routes *Routes | ||
| 112 | } | 134 | } | ||
| @@ -132,8 +154,24 @@ func New(cfg Config) *Adapter { | |||||
| 132 | if refreshEvery == 0 { | 154 | if refreshEvery == 0 { | ||
| 133 | refreshEvery = DefaultRefreshEvery | 155 | refreshEvery = DefaultRefreshEvery | ||
| 134 | } | 156 | } | ||
| 157 | + | idleTTL := cfg.IdleTTL | |||
| 158 | + | if idleTTL == 0 { | |||
| 159 | + | idleTTL = DefaultIdleTTL | |||
| 160 | + | } | |||
| 161 | + | deltaCache := cfg.DeltaCacheSize | |||
| 162 | + | if deltaCache == 0 { | |||
| 163 | + | deltaCache = DefaultDeltaCacheSize | |||
| 164 | + | } | |||
| 165 | + | maxObject := cfg.MaxObjectSize | |||
| 166 | + | if maxObject == 0 { | |||
| 167 | + | maxObject = DefaultMaxObjectSize | |||
| 168 | + | } | |||
| 135 | a := &Adapter{cli: cfg.Fallback, dirOf: cfg.RepoDir, routes: routes} | 169 | a := &Adapter{cli: cfg.Fallback, dirOf: cfg.RepoDir, routes: routes} | ||
| 136 | - | a.cache = newCache(cfg.MaxRepos, refreshEvery, a.logUnsupported) | 170 | + | a.cache = newCache(cacheConfig{ |
| 171 | + | max: cfg.MaxRepos, refreshEvery: refreshEvery, idleTTL: idleTTL, | |||
| 172 | + | openOpts: []git.Option{git.WithDeltaCacheSize(deltaCache), git.WithMaxObjectSize(maxObject)}, | |||
| 173 | + | onUnsupport: a.logUnsupported, | |||
| 174 | + | }) | |||
| 137 | return a | 175 | return a | ||
| 138 | } | 176 | } | ||
| 139 | 177 | ||||
| @@ -148,6 +186,17 @@ func (a *Adapter) Forget(dir string) { | |||||
| 148 | a.logged.Delete(dir) | 186 | a.logged.Delete(dir) | ||
| 149 | } | 187 | } | ||
| 150 | 188 | ||||
| 189 | + | // Refresh tells the adapter that ermite wrote to the repository (a push, the | |||
| 190 | + | // maintenance): the library refreshes it now, if it is open, instead of at the | |||
| 191 | + | // next read after RefreshEvery. A repository that is not open has nothing to | |||
| 192 | + | // refresh. A failure is logged, not returned: reads do not depend on it. | |||
| 193 | + | func (a *Adapter) Refresh(ctx context.Context, repo domain.RepoName) { | |||
| 194 | + | if err := a.cache.refreshNow(ctx, repo, a.dirOf(repo)); err != nil { | |||
| 195 | + | log := zerowrap.FromCtx(ctx) | |||
| 196 | + | log.Warn().Err(err).Str("repo", repo.String()).Msg("refresh of the open repository failed") | |||
| 197 | + | } | |||
| 198 | + | } | |||
| 199 | + | ||||
| 151 | // OpenRepos returns the number of repositories in the cache, for tests. | 200 | // OpenRepos returns the number of repositories in the cache, for tests. | ||
| 152 | func (a *Adapter) OpenRepos() int { return a.cache.size() } | 201 | func (a *Adapter) OpenRepos() int { return a.cache.size() } | ||
| 153 | 202 | ||||