fix(web): bound the concurrent path list builds
A build walks a whole tree and holds the list in memory, and each commit had its own, so many distinct commits could build at once. The cache now has four build slots shared by all keys: a caller that would start a build waits for a slot or for its context, and the build keeps the slot until it ends, whether its callers stay or not. The cache keeps one rule for the size of an entry: a list over a quarter of the budget is not stored. The separate entry maximum and the unused get and put are gone.
- commit
- c75c5f3bcdccb052d002a2d36c2e028388253f4a
- author
- brice · · 2026-10-12 05:10 UTC
- parent
- b5e1b0c
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +229 -81
M internal/adapters/web/pathcache.go +74 -36
| @@ -14,9 +14,10 @@ import ( | |||
| 14 | 14 | const ( | |
| 15 | 15 | // pathCacheBytes is the budget of the cache of path lists. | |
| 16 | 16 | pathCacheBytes = 64 << 20 | |
| 17 | - | // pathEntryMax is the largest list the cache keeps; a bigger one is built | |
| 18 | - | // again for each request that needs it. | |
| 19 | - | pathEntryMax = 16 << 20 | |
| 17 | + | // pathBuilds is how many path lists are built at the same time, for all | |
| 18 | + | // commits and repositories: a build walks a whole tree and holds the | |
| 19 | + | // list in memory. | |
| 20 | + | pathBuilds = 4 | |
| 20 | 21 | ) | |
| 21 | 22 | ||
| 22 | 23 | // pathCache keeps the gzipped JSON path lists of commits, least recently used | |
| @@ -24,15 +25,19 @@ const ( | |||
| 24 | 25 | // entry is right for every repository and every visitor; who may read it is | |
| 25 | 26 | // decided before the cache is asked (Server.paths). | |
| 26 | 27 | // | |
| 27 | - | // Concurrent misses for one commit of one repository share a single build. | |
| 28 | + | // Concurrent misses for one commit of one repository share a single build, | |
| 29 | + | // and only a few builds run at a time. | |
| 30 | + | // | |
| 31 | + | // A list over a quarter of the budget is not kept (one big list would push | |
| 32 | + | // out most of the others): it is built again for each request that needs it. | |
| 28 | 33 | type pathCache struct { | |
| 29 | 34 | mu sync.Mutex | |
| 30 | 35 | budget int | |
| 31 | 36 | size int | |
| 32 | - | max int // largest entry kept | |
| 33 | 37 | lru *list.List // of *pathEntry, most recently used first | |
| 34 | 38 | byID map[string]*list.Element | |
| 35 | 39 | building map[string]*pathBuild // by repository and commit | |
| 40 | + | slots chan struct{} // one token per build running | |
| 36 | 41 | } | |
| 37 | 42 | ||
| 38 | 43 | type pathEntry struct { | |
| @@ -42,25 +47,20 @@ type pathEntry struct { | |||
| 42 | 47 | ||
| 43 | 48 | // pathBuild is a build in progress; done is closed when body and err are set. | |
| 44 | 49 | type pathBuild struct { | |
| 45 | - | done chan struct{} | |
| 46 | - | body []byte | |
| 47 | - | err error | |
| 50 | + | waiters int // callers that joined it, for the tests; guarded by pathCache.mu | |
| 51 | + | done chan struct{} | |
| 52 | + | body []byte | |
| 53 | + | err error | |
| 48 | 54 | } | |
| 49 | 55 | ||
| 50 | - | func newPathCache(budget, max int) *pathCache { | |
| 56 | + | func newPathCache(budget, builds int) *pathCache { | |
| 51 | 57 | return &pathCache{ | |
| 52 | - | budget: budget, max: max, lru: list.New(), | |
| 58 | + | budget: budget, lru: list.New(), | |
| 53 | 59 | byID: map[string]*list.Element{}, building: map[string]*pathBuild{}, | |
| 60 | + | slots: make(chan struct{}, builds), | |
| 54 | 61 | } | |
| 55 | 62 | } | |
| 56 | 63 | ||
| 57 | - | // get returns the list of a commit and marks it as used. | |
| 58 | - | func (c *pathCache) get(id string) ([]byte, bool) { | |
| 59 | - | c.mu.Lock() | |
| 60 | - | defer c.mu.Unlock() | |
| 61 | - | return c.getLocked(id) | |
| 62 | - | } | |
| 63 | - | ||
| 64 | 64 | func (c *pathCache) getLocked(id string) ([]byte, bool) { | |
| 65 | 65 | el, ok := c.byID[id] | |
| 66 | 66 | if !ok { | |
| @@ -70,16 +70,10 @@ func (c *pathCache) getLocked(id string) ([]byte, bool) { | |||
| 70 | 70 | return el.Value.(*pathEntry).body, true | |
| 71 | 71 | } | |
| 72 | 72 | ||
| 73 | - | // put stores a list, evicting the least recently used ones to stay in budget. | |
| 74 | - | // A list over the entry limit or the budget is not stored. | |
| 75 | - | func (c *pathCache) put(id string, body []byte) { | |
| 76 | - | c.mu.Lock() | |
| 77 | - | defer c.mu.Unlock() | |
| 78 | - | c.putLocked(id, body) | |
| 79 | - | } | |
| 80 | - | ||
| 73 | + | // putLocked stores a list, evicting the least recently used ones to stay in | |
| 74 | + | // budget. A list over a quarter of the budget is not stored. | |
| 81 | 75 | func (c *pathCache) putLocked(id string, body []byte) { | |
| 82 | - | if len(body) > c.max || len(body) > c.budget { | |
| 76 | + | if len(body) > c.budget/4 { | |
| 83 | 77 | return | |
| 84 | 78 | } | |
| 85 | 79 | if el, ok := c.byID[id]; ok { | |
| @@ -105,24 +99,32 @@ func (c *pathCache) putLocked(id string, body []byte) { | |||
| 105 | 99 | // build runs in its own goroutine, so no caller is held by it: each one, | |
| 106 | 100 | // the one that started it included, waits for the build or for its own ctx. | |
| 107 | 101 | // | |
| 102 | + | // Only pathBuilds builds run at a time, whatever their commits: a caller that | |
| 103 | + | // would start one waits for a free slot, or for its ctx. The build keeps the | |
| 104 | + | // slot until it ends, even if every caller has left. | |
| 105 | + | // | |
| 108 | 106 | // Builds in progress are keyed by repository and commit, though the cache is | |
| 109 | 107 | // keyed by commit only: a failure of one repository (the commit is missing | |
| 110 | 108 | // there, its access rule changed) must not be handed to the requests of | |
| 111 | 109 | // another one that would have succeeded. | |
| 112 | 110 | func (c *pathCache) load(ctx context.Context, repo, id string, build func() ([]byte, error)) ([]byte, error) { | |
| 113 | 111 | key := repo + "\x00" + id // a repository name has no NUL | |
| 114 | - | c.mu.Lock() | |
| 115 | - | if b, ok := c.getLocked(id); ok { | |
| 116 | - | c.mu.Unlock() | |
| 117 | - | return b, nil | |
| 112 | + | f, body, ok := c.join(key, id) | |
| 113 | + | if ok { | |
| 114 | + | return body, nil | |
| 118 | 115 | } | |
| 119 | - | f, ok := c.building[key] | |
| 120 | - | if !ok { | |
| 121 | - | f = &pathBuild{done: make(chan struct{})} | |
| 122 | - | c.building[key] = f | |
| 123 | - | go c.run(ctx, key, id, f, build) | |
| 116 | + | if f == nil { | |
| 117 | + | // Nothing to join: take a slot, then look again, since the commit | |
| 118 | + | // may have been built while this caller waited. | |
| 119 | + | select { | |
| 120 | + | case c.slots <- struct{}{}: | |
| 121 | + | case <-ctx.Done(): | |
| 122 | + | return nil, ctx.Err() | |
| 123 | + | } | |
| 124 | + | if f, body, ok = c.joinOrStart(ctx, key, id, build); ok { | |
| 125 | + | return body, nil | |
| 126 | + | } | |
| 124 | 127 | } | |
| 125 | - | c.mu.Unlock() | |
| 126 | 128 | select { | |
| 127 | 129 | case <-f.done: | |
| 128 | 130 | return f.body, f.err | |
| @@ -131,11 +133,47 @@ func (c *pathCache) load(ctx context.Context, repo, id string, build func() ([]b | |||
| 131 | 133 | } | |
| 132 | 134 | } | |
| 133 | 135 | ||
| 136 | + | // join returns the list of a commit if it is cached, or the build in progress | |
| 137 | + | // for it. | |
| 138 | + | func (c *pathCache) join(key, id string) (f *pathBuild, body []byte, ok bool) { | |
| 139 | + | c.mu.Lock() | |
| 140 | + | defer c.mu.Unlock() | |
| 141 | + | if b, ok := c.getLocked(id); ok { | |
| 142 | + | return nil, b, true | |
| 143 | + | } | |
| 144 | + | f = c.building[key] | |
| 145 | + | if f != nil { | |
| 146 | + | f.waiters++ | |
| 147 | + | } | |
| 148 | + | return f, nil, false | |
| 149 | + | } | |
| 150 | + | ||
| 151 | + | // joinOrStart is join for a caller that holds a slot: it starts the build when | |
| 152 | + | // there is still none, and gives the slot back when it does not need it. | |
| 153 | + | func (c *pathCache) joinOrStart(ctx context.Context, key, id string, build func() ([]byte, error)) (f *pathBuild, body []byte, ok bool) { | |
| 154 | + | c.mu.Lock() | |
| 155 | + | defer c.mu.Unlock() | |
| 156 | + | if b, ok := c.getLocked(id); ok { | |
| 157 | + | <-c.slots | |
| 158 | + | return nil, b, true | |
| 159 | + | } | |
| 160 | + | if f, ok := c.building[key]; ok { | |
| 161 | + | <-c.slots | |
| 162 | + | f.waiters++ | |
| 163 | + | return f, nil, false | |
| 164 | + | } | |
| 165 | + | f = &pathBuild{done: make(chan struct{}), waiters: 1} | |
| 166 | + | c.building[key] = f | |
| 167 | + | go c.run(ctx, key, id, f, build) | |
| 168 | + | return f, nil, false | |
| 169 | + | } | |
| 170 | + | ||
| 134 | 171 | // run executes a build and publishes it. A panic of build is logged and | |
| 135 | 172 | // handed to the waiters as errBuildPanicked: it cannot propagate, since | |
| 136 | 173 | // nobody above this goroutine could recover it and it would end the process. | |
| 137 | 174 | func (c *pathCache) run(ctx context.Context, key, id string, f *pathBuild, build func() ([]byte, error)) { | |
| 138 | 175 | defer func() { | |
| 176 | + | defer func() { <-c.slots }() | |
| 139 | 177 | if r := recover(); r != nil { | |
| 140 | 178 | f.body, f.err = nil, errBuildPanicked | |
| 141 | 179 | zerowrap.Ctx(ctx).Error().Interface("panic", r).Str("stack", string(debug.Stack())).Msg("path list build panicked") | |
| @@ -14,9 +14,10 @@ import ( | |||||
| 14 | const ( | 14 | const ( | ||
| 15 | // pathCacheBytes is the budget of the cache of path lists. | 15 | // pathCacheBytes is the budget of the cache of path lists. | ||
| 16 | pathCacheBytes = 64 << 20 | 16 | pathCacheBytes = 64 << 20 | ||
| 17 | - | // pathEntryMax is the largest list the cache keeps; a bigger one is built | 17 | + | // pathBuilds is how many path lists are built at the same time, for all |
| 18 | - | // again for each request that needs it. | 18 | + | // commits and repositories: a build walks a whole tree and holds the |
| 19 | - | pathEntryMax = 16 << 20 | 19 | + | // list in memory. |
| 20 | + | pathBuilds = 4 | |||
| 20 | ) | 21 | ) | ||
| 21 | 22 | ||||
| 22 | // pathCache keeps the gzipped JSON path lists of commits, least recently used | 23 | // pathCache keeps the gzipped JSON path lists of commits, least recently used | ||
| @@ -24,15 +25,19 @@ const ( | |||||
| 24 | // entry is right for every repository and every visitor; who may read it is | 25 | // entry is right for every repository and every visitor; who may read it is | ||
| 25 | // decided before the cache is asked (Server.paths). | 26 | // decided before the cache is asked (Server.paths). | ||
| 26 | // | 27 | // | ||
| 27 | - | // Concurrent misses for one commit of one repository share a single build. | 28 | + | // Concurrent misses for one commit of one repository share a single build, |
| 29 | + | // and only a few builds run at a time. | |||
| 30 | + | // | |||
| 31 | + | // A list over a quarter of the budget is not kept (one big list would push | |||
| 32 | + | // out most of the others): it is built again for each request that needs it. | |||
| 28 | type pathCache struct { | 33 | type pathCache struct { | ||
| 29 | mu sync.Mutex | 34 | mu sync.Mutex | ||
| 30 | budget int | 35 | budget int | ||
| 31 | size int | 36 | size int | ||
| 32 | - | max int // largest entry kept | |||
| 33 | lru *list.List // of *pathEntry, most recently used first | 37 | lru *list.List // of *pathEntry, most recently used first | ||
| 34 | byID map[string]*list.Element | 38 | byID map[string]*list.Element | ||
| 35 | building map[string]*pathBuild // by repository and commit | 39 | building map[string]*pathBuild // by repository and commit | ||
| 40 | + | slots chan struct{} // one token per build running | |||
| 36 | } | 41 | } | ||
| 37 | 42 | ||||
| 38 | type pathEntry struct { | 43 | type pathEntry struct { | ||
| @@ -42,25 +47,20 @@ type pathEntry struct { | |||||
| 42 | 47 | ||||
| 43 | // pathBuild is a build in progress; done is closed when body and err are set. | 48 | // pathBuild is a build in progress; done is closed when body and err are set. | ||
| 44 | type pathBuild struct { | 49 | type pathBuild struct { | ||
| 45 | - | done chan struct{} | 50 | + | waiters int // callers that joined it, for the tests; guarded by pathCache.mu |
| 46 | - | body []byte | 51 | + | done chan struct{} |
| 47 | - | err error | 52 | + | body []byte |
| 53 | + | err error | |||
| 48 | } | 54 | } | ||
| 49 | 55 | ||||
| 50 | - | func newPathCache(budget, max int) *pathCache { | 56 | + | func newPathCache(budget, builds int) *pathCache { |
| 51 | return &pathCache{ | 57 | return &pathCache{ | ||
| 52 | - | budget: budget, max: max, lru: list.New(), | 58 | + | budget: budget, lru: list.New(), |
| 53 | byID: map[string]*list.Element{}, building: map[string]*pathBuild{}, | 59 | byID: map[string]*list.Element{}, building: map[string]*pathBuild{}, | ||
| 60 | + | slots: make(chan struct{}, builds), | |||
| 54 | } | 61 | } | ||
| 55 | } | 62 | } | ||
| 56 | 63 | ||||
| 57 | - | // get returns the list of a commit and marks it as used. | |||
| 58 | - | func (c *pathCache) get(id string) ([]byte, bool) { | |||
| 59 | - | c.mu.Lock() | |||
| 60 | - | defer c.mu.Unlock() | |||
| 61 | - | return c.getLocked(id) | |||
| 62 | - | } | |||
| 63 | - | ||||
| 64 | func (c *pathCache) getLocked(id string) ([]byte, bool) { | 64 | func (c *pathCache) getLocked(id string) ([]byte, bool) { | ||
| 65 | el, ok := c.byID[id] | 65 | el, ok := c.byID[id] | ||
| 66 | if !ok { | 66 | if !ok { | ||
| @@ -70,16 +70,10 @@ func (c *pathCache) getLocked(id string) ([]byte, bool) { | |||||
| 70 | return el.Value.(*pathEntry).body, true | 70 | return el.Value.(*pathEntry).body, true | ||
| 71 | } | 71 | } | ||
| 72 | 72 | ||||
| 73 | - | // put stores a list, evicting the least recently used ones to stay in budget. | 73 | + | // putLocked stores a list, evicting the least recently used ones to stay in |
| 74 | - | // A list over the entry limit or the budget is not stored. | 74 | + | // budget. A list over a quarter of the budget is not stored. |
| 75 | - | func (c *pathCache) put(id string, body []byte) { | |||
| 76 | - | c.mu.Lock() | |||
| 77 | - | defer c.mu.Unlock() | |||
| 78 | - | c.putLocked(id, body) | |||
| 79 | - | } | |||
| 80 | - | ||||
| 81 | func (c *pathCache) putLocked(id string, body []byte) { | 75 | func (c *pathCache) putLocked(id string, body []byte) { | ||
| 82 | - | if len(body) > c.max || len(body) > c.budget { | 76 | + | if len(body) > c.budget/4 { |
| 83 | return | 77 | return | ||
| 84 | } | 78 | } | ||
| 85 | if el, ok := c.byID[id]; ok { | 79 | if el, ok := c.byID[id]; ok { | ||
| @@ -105,24 +99,32 @@ func (c *pathCache) putLocked(id string, body []byte) { | |||||
| 105 | // build runs in its own goroutine, so no caller is held by it: each one, | 99 | // build runs in its own goroutine, so no caller is held by it: each one, | ||
| 106 | // the one that started it included, waits for the build or for its own ctx. | 100 | // the one that started it included, waits for the build or for its own ctx. | ||
| 107 | // | 101 | // | ||
| 102 | + | // Only pathBuilds builds run at a time, whatever their commits: a caller that | |||
| 103 | + | // would start one waits for a free slot, or for its ctx. The build keeps the | |||
| 104 | + | // slot until it ends, even if every caller has left. | |||
| 105 | + | // | |||
| 108 | // Builds in progress are keyed by repository and commit, though the cache is | 106 | // Builds in progress are keyed by repository and commit, though the cache is | ||
| 109 | // keyed by commit only: a failure of one repository (the commit is missing | 107 | // keyed by commit only: a failure of one repository (the commit is missing | ||
| 110 | // there, its access rule changed) must not be handed to the requests of | 108 | // there, its access rule changed) must not be handed to the requests of | ||
| 111 | // another one that would have succeeded. | 109 | // another one that would have succeeded. | ||
| 112 | func (c *pathCache) load(ctx context.Context, repo, id string, build func() ([]byte, error)) ([]byte, error) { | 110 | func (c *pathCache) load(ctx context.Context, repo, id string, build func() ([]byte, error)) ([]byte, error) { | ||
| 113 | key := repo + "\x00" + id // a repository name has no NUL | 111 | key := repo + "\x00" + id // a repository name has no NUL | ||
| 114 | - | c.mu.Lock() | 112 | + | f, body, ok := c.join(key, id) |
| 115 | - | if b, ok := c.getLocked(id); ok { | 113 | + | if ok { |
| 116 | - | c.mu.Unlock() | 114 | + | return body, nil |
| 117 | - | return b, nil | |||
| 118 | } | 115 | } | ||
| 119 | - | f, ok := c.building[key] | 116 | + | if f == nil { |
| 120 | - | if !ok { | 117 | + | // Nothing to join: take a slot, then look again, since the commit |
| 121 | - | f = &pathBuild{done: make(chan struct{})} | 118 | + | // may have been built while this caller waited. |
| 122 | - | c.building[key] = f | 119 | + | select { |
| 123 | - | go c.run(ctx, key, id, f, build) | 120 | + | case c.slots <- struct{}{}: |
| 121 | + | case <-ctx.Done(): | |||
| 122 | + | return nil, ctx.Err() | |||
| 123 | + | } | |||
| 124 | + | if f, body, ok = c.joinOrStart(ctx, key, id, build); ok { | |||
| 125 | + | return body, nil | |||
| 126 | + | } | |||
| 124 | } | 127 | } | ||
| 125 | - | c.mu.Unlock() | |||
| 126 | select { | 128 | select { | ||
| 127 | case <-f.done: | 129 | case <-f.done: | ||
| 128 | return f.body, f.err | 130 | return f.body, f.err | ||
| @@ -131,11 +133,47 @@ func (c *pathCache) load(ctx context.Context, repo, id string, build func() ([]b | |||||
| 131 | } | 133 | } | ||
| 132 | } | 134 | } | ||
| 133 | 135 | ||||
| 136 | + | // join returns the list of a commit if it is cached, or the build in progress | |||
| 137 | + | // for it. | |||
| 138 | + | func (c *pathCache) join(key, id string) (f *pathBuild, body []byte, ok bool) { | |||
| 139 | + | c.mu.Lock() | |||
| 140 | + | defer c.mu.Unlock() | |||
| 141 | + | if b, ok := c.getLocked(id); ok { | |||
| 142 | + | return nil, b, true | |||
| 143 | + | } | |||
| 144 | + | f = c.building[key] | |||
| 145 | + | if f != nil { | |||
| 146 | + | f.waiters++ | |||
| 147 | + | } | |||
| 148 | + | return f, nil, false | |||
| 149 | + | } | |||
| 150 | + | ||||
| 151 | + | // joinOrStart is join for a caller that holds a slot: it starts the build when | |||
| 152 | + | // there is still none, and gives the slot back when it does not need it. | |||
| 153 | + | func (c *pathCache) joinOrStart(ctx context.Context, key, id string, build func() ([]byte, error)) (f *pathBuild, body []byte, ok bool) { | |||
| 154 | + | c.mu.Lock() | |||
| 155 | + | defer c.mu.Unlock() | |||
| 156 | + | if b, ok := c.getLocked(id); ok { | |||
| 157 | + | <-c.slots | |||
| 158 | + | return nil, b, true | |||
| 159 | + | } | |||
| 160 | + | if f, ok := c.building[key]; ok { | |||
| 161 | + | <-c.slots | |||
| 162 | + | f.waiters++ | |||
| 163 | + | return f, nil, false | |||
| 164 | + | } | |||
| 165 | + | f = &pathBuild{done: make(chan struct{}), waiters: 1} | |||
| 166 | + | c.building[key] = f | |||
| 167 | + | go c.run(ctx, key, id, f, build) | |||
| 168 | + | return f, nil, false | |||
| 169 | + | } | |||
| 170 | + | ||||
| 134 | // run executes a build and publishes it. A panic of build is logged and | 171 | // run executes a build and publishes it. A panic of build is logged and | ||
| 135 | // handed to the waiters as errBuildPanicked: it cannot propagate, since | 172 | // handed to the waiters as errBuildPanicked: it cannot propagate, since | ||
| 136 | // nobody above this goroutine could recover it and it would end the process. | 173 | // nobody above this goroutine could recover it and it would end the process. | ||
| 137 | func (c *pathCache) run(ctx context.Context, key, id string, f *pathBuild, build func() ([]byte, error)) { | 174 | func (c *pathCache) run(ctx context.Context, key, id string, f *pathBuild, build func() ([]byte, error)) { | ||
| 138 | defer func() { | 175 | defer func() { | ||
| 176 | + | defer func() { <-c.slots }() | |||
| 139 | if r := recover(); r != nil { | 177 | if r := recover(); r != nil { | ||
| 140 | f.body, f.err = nil, errBuildPanicked | 178 | f.body, f.err = nil, errBuildPanicked | ||
| 141 | zerowrap.Ctx(ctx).Error().Interface("panic", r).Str("stack", string(debug.Stack())).Msg("path list build panicked") | 179 | zerowrap.Ctx(ctx).Error().Interface("panic", r).Str("stack", string(debug.Stack())).Msg("path list build panicked") | ||
M internal/adapters/web/pathcache_test.go +154 -44
| @@ -15,55 +15,80 @@ import ( | |||
| 15 | 15 | "github.com/stretchr/testify/require" | |
| 16 | 16 | ) | |
| 17 | 17 | ||
| 18 | + | // store puts a list in the cache the way a finished build does. | |
| 19 | + | func (c *pathCache) store(id string, body []byte) { | |
| 20 | + | c.mu.Lock() | |
| 21 | + | defer c.mu.Unlock() | |
| 22 | + | c.putLocked(id, body) | |
| 23 | + | } | |
| 24 | + | ||
| 25 | + | // has says whether a list is cached, marking it as used. | |
| 26 | + | func (c *pathCache) has(id string) bool { | |
| 27 | + | c.mu.Lock() | |
| 28 | + | defer c.mu.Unlock() | |
| 29 | + | _, ok := c.getLocked(id) | |
| 30 | + | return ok | |
| 31 | + | } | |
| 32 | + | ||
| 18 | 33 | func TestPathCacheEvictsLeastRecentlyUsed(t *testing.T) { | |
| 19 | - | c := newPathCache(100, 60) | |
| 34 | + | c := newPathCache(100, 1) | |
| 20 | 35 | blob := func(n int) []byte { return make([]byte, n) } | |
| 21 | - | c.put("a", blob(40)) | |
| 22 | - | c.put("b", blob(40)) | |
| 23 | - | _, ok := c.get("a") // a is now the most recent | |
| 24 | - | require.True(t, ok) | |
| 25 | - | c.put("c", blob(40)) // 120 > 100: b, the least recent, goes | |
| 26 | - | _, ok = c.get("b") | |
| 27 | - | assert.False(t, ok, "b was evicted") | |
| 28 | - | _, okA := c.get("a") | |
| 29 | - | _, okC := c.get("c") | |
| 30 | - | assert.True(t, okA && okC) | |
| 31 | - | assert.Equal(t, 80, c.size) | |
| 36 | + | c.store("a", blob(25)) | |
| 37 | + | c.store("b", blob(25)) | |
| 38 | + | c.store("e", blob(25)) | |
| 39 | + | require.True(t, c.has("a")) // a is now the most recent | |
| 40 | + | c.store("c", blob(25)) // 100 fit | |
| 41 | + | c.store("d", blob(10)) // 110 > 100: b, the least recent, goes | |
| 42 | + | assert.False(t, c.has("b"), "b was evicted") | |
| 43 | + | assert.True(t, c.has("a") && c.has("c") && c.has("d") && c.has("e")) | |
| 44 | + | assert.Equal(t, 85, c.size) | |
| 32 | 45 | ||
| 33 | 46 | // Replacing an entry accounts for the old size. | |
| 34 | - | c.put("a", blob(10)) | |
| 35 | - | assert.Equal(t, 50, c.size) | |
| 47 | + | c.store("a", blob(10)) | |
| 48 | + | assert.Equal(t, 70, c.size) | |
| 36 | 49 | ||
| 37 | 50 | // A big one pushes out as many as needed. | |
| 38 | - | c.put("d", blob(60)) | |
| 51 | + | c.store("f", blob(25)) | |
| 52 | + | c.store("g", blob(25)) | |
| 39 | 53 | assert.LessOrEqual(t, c.size, 100) | |
| 40 | - | _, ok = c.get("d") | |
| 41 | - | assert.True(t, ok) | |
| 54 | + | assert.True(t, c.has("g")) | |
| 42 | 55 | } | |
| 43 | 56 | ||
| 57 | + | // A list over a quarter of the budget is not kept, since it would push out the | |
| 58 | + | // others; one at the limit is. | |
| 44 | 59 | func TestPathCacheSkipsOversizedEntries(t *testing.T) { | |
| 45 | - | c := newPathCache(100, 60) | |
| 46 | - | c.put("big", make([]byte, 61)) | |
| 47 | - | _, ok := c.get("big") | |
| 48 | - | assert.False(t, ok, "over the entry limit") | |
| 49 | - | c = newPathCache(50, 60) | |
| 50 | - | c.put("big", make([]byte, 51)) | |
| 51 | - | _, ok = c.get("big") | |
| 52 | - | assert.False(t, ok, "over the budget") | |
| 60 | + | c := newPathCache(100, 1) | |
| 61 | + | var builds atomic.Int32 | |
| 62 | + | big := func(n int) func() ([]byte, error) { | |
| 63 | + | return func() ([]byte, error) { builds.Add(1); return make([]byte, n), nil } | |
| 64 | + | } | |
| 65 | + | for range 2 { | |
| 66 | + | b, err := c.load(context.Background(), "r", "big", big(26)) | |
| 67 | + | require.NoError(t, err) | |
| 68 | + | assert.Len(t, b, 26, "the caller still gets the list") | |
| 69 | + | } | |
| 70 | + | assert.EqualValues(t, 2, builds.Load(), "not cached, so built again") | |
| 71 | + | assert.False(t, c.has("big")) | |
| 53 | 72 | assert.Zero(t, c.size) | |
| 73 | + | ||
| 74 | + | for range 2 { | |
| 75 | + | _, err := c.load(context.Background(), "r", "fits", big(25)) | |
| 76 | + | require.NoError(t, err) | |
| 77 | + | } | |
| 78 | + | assert.EqualValues(t, 3, builds.Load(), "at a quarter: kept") | |
| 79 | + | assert.Equal(t, 25, c.size) | |
| 54 | 80 | } | |
| 55 | 81 | ||
| 56 | 82 | func TestPathCacheConcurrent(t *testing.T) { | |
| 57 | - | c := newPathCache(1000, 200) | |
| 83 | + | c := newPathCache(1000, 1) | |
| 58 | 84 | var wg sync.WaitGroup | |
| 59 | 85 | for g := range 8 { | |
| 60 | 86 | wg.Add(1) | |
| 61 | 87 | go func() { | |
| 62 | 88 | defer wg.Done() | |
| 63 | 89 | for i := range 500 { | |
| 64 | - | id := fmt.Sprint((g + i) % 17) | |
| 65 | - | c.put(id, make([]byte, 10+i%90)) | |
| 66 | - | c.get(fmt.Sprint(i % 17)) | |
| 90 | + | c.store(fmt.Sprint((g+i)%17), make([]byte, 10+i%90)) | |
| 91 | + | c.has(fmt.Sprint(i % 17)) | |
| 67 | 92 | } | |
| 68 | 93 | }() | |
| 69 | 94 | } | |
| @@ -77,10 +102,27 @@ func TestPathCacheConcurrent(t *testing.T) { | |||
| 77 | 102 | assert.Equal(t, len(c.byID), c.lru.Len()) | |
| 78 | 103 | } | |
| 79 | 104 | ||
| 105 | + | // waiters is the number of callers waiting for the build of a key, -1 when | |
| 106 | + | // none is in flight. | |
| 107 | + | func waiters(c *pathCache, repo, id string) int { | |
| 108 | + | c.mu.Lock() | |
| 109 | + | defer c.mu.Unlock() | |
| 110 | + | if f := c.building[repo+"\x00"+id]; f != nil { | |
| 111 | + | return f.waiters | |
| 112 | + | } | |
| 113 | + | return -1 | |
| 114 | + | } | |
| 115 | + | ||
| 116 | + | func building(c *pathCache, n int) bool { | |
| 117 | + | c.mu.Lock() | |
| 118 | + | defer c.mu.Unlock() | |
| 119 | + | return len(c.building) == n | |
| 120 | + | } | |
| 121 | + | ||
| 80 | 122 | // Concurrent misses for one commit build once; a failure is shared and not | |
| 81 | 123 | // stored; a waiter that gives up leaves the build running. | |
| 82 | 124 | func TestPathCacheLoadBuildsOnce(t *testing.T) { | |
| 83 | - | c := newPathCache(1000, 200) | |
| 125 | + | c := newPathCache(1000, 4) | |
| 84 | 126 | var builds atomic.Int32 | |
| 85 | 127 | release := make(chan struct{}) | |
| 86 | 128 | build := func() ([]byte, error) { | |
| @@ -88,8 +130,9 @@ func TestPathCacheLoadBuildsOnce(t *testing.T) { | |||
| 88 | 130 | <-release | |
| 89 | 131 | return []byte("list"), nil | |
| 90 | 132 | } | |
| 133 | + | const callers = 10 | |
| 91 | 134 | var wg sync.WaitGroup | |
| 92 | - | got := make([][]byte, 10) | |
| 135 | + | got := make([][]byte, callers) | |
| 93 | 136 | for i := range got { | |
| 94 | 137 | wg.Add(1) | |
| 95 | 138 | go func() { | |
| @@ -99,8 +142,10 @@ func TestPathCacheLoadBuildsOnce(t *testing.T) { | |||
| 99 | 142 | got[i] = b | |
| 100 | 143 | }() | |
| 101 | 144 | } | |
| 145 | + | // Every caller waits for the one build before it is released: none can | |
| 146 | + | // find the list cached, so one build serving all ten shows they joined it. | |
| 147 | + | require.Eventually(t, func() bool { return waiters(c, "r", "x") == callers }, 5*time.Second, time.Millisecond) | |
| 102 | 148 | // A waiter whose context ends does not stop the others. | |
| 103 | - | require.Eventually(t, func() bool { c.mu.Lock(); defer c.mu.Unlock(); return c.building["r\x00x"] != nil }, time.Second, time.Millisecond) | |
| 104 | 149 | ctx, cancel := context.WithCancel(context.Background()) | |
| 105 | 150 | cancel() | |
| 106 | 151 | _, err := c.load(ctx, "r", "x", build) | |
| @@ -111,25 +156,91 @@ func TestPathCacheLoadBuildsOnce(t *testing.T) { | |||
| 111 | 156 | for _, b := range got { | |
| 112 | 157 | assert.Equal(t, "list", string(b)) | |
| 113 | 158 | } | |
| 114 | - | b, ok := c.get("x") | |
| 115 | - | assert.True(t, ok) | |
| 116 | - | assert.Equal(t, "list", string(b)) | |
| 159 | + | assert.True(t, c.has("x")) | |
| 117 | 160 | ||
| 118 | 161 | boom := errors.New("boom") | |
| 119 | 162 | _, err = c.load(context.Background(), "r", "y", func() ([]byte, error) { return nil, boom }) | |
| 120 | 163 | assert.ErrorIs(t, err, boom) | |
| 121 | - | _, ok = c.get("y") | |
| 122 | - | assert.False(t, ok, "a failure is not cached") | |
| 123 | - | b, err = c.load(context.Background(), "r", "y", func() ([]byte, error) { return []byte("ok"), nil }) | |
| 164 | + | assert.False(t, c.has("y"), "a failure is not cached") | |
| 165 | + | b, err := c.load(context.Background(), "r", "y", func() ([]byte, error) { return []byte("ok"), nil }) | |
| 124 | 166 | require.NoError(t, err) | |
| 125 | 167 | assert.Equal(t, "ok", string(b), "the next request builds again") | |
| 126 | 168 | assert.Empty(t, c.building) | |
| 127 | 169 | } | |
| 128 | 170 | ||
| 171 | + | // The builds of all commits share the slots: with one slot, a load of another | |
| 172 | + | // commit waits for the running build, and leaves with its ctx if it is | |
| 173 | + | // cancelled first. The slot is the build's, not the caller's. | |
| 174 | + | func TestPathCacheLoadBoundsBuilds(t *testing.T) { | |
| 175 | + | c := newPathCache(1000, 1) | |
| 176 | + | started, release := make(chan struct{}), make(chan struct{}) | |
| 177 | + | first := make(chan error, 1) | |
| 178 | + | go func() { | |
| 179 | + | _, err := c.load(context.Background(), "r", "a", func() ([]byte, error) { close(started); <-release; return []byte("a"), nil }) | |
| 180 | + | first <- err | |
| 181 | + | }() | |
| 182 | + | <-started | |
| 183 | + | ||
| 184 | + | // A second commit waits for the slot... | |
| 185 | + | var secondBuilt atomic.Bool | |
| 186 | + | second := make(chan error, 1) | |
| 187 | + | go func() { | |
| 188 | + | _, err := c.load(context.Background(), "r", "b", func() ([]byte, error) { secondBuilt.Store(true); return []byte("b"), nil }) | |
| 189 | + | second <- err | |
| 190 | + | }() | |
| 191 | + | time.Sleep(30 * time.Millisecond) | |
| 192 | + | assert.False(t, secondBuilt.Load(), "no slot, no build") | |
| 193 | + | select { | |
| 194 | + | case <-second: | |
| 195 | + | t.Fatal("the second load did not wait for a slot") | |
| 196 | + | default: | |
| 197 | + | } | |
| 198 | + | ||
| 199 | + | // ...or leaves with its ctx. | |
| 200 | + | ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) | |
| 201 | + | defer cancel() | |
| 202 | + | _, err := c.load(ctx, "r", "c", func() ([]byte, error) { t.Error("built without a slot"); return nil, nil }) | |
| 203 | + | assert.ErrorIs(t, err, context.DeadlineExceeded) | |
| 204 | + | ||
| 205 | + | // A caller of the running build is not held by the slots. | |
| 206 | + | _, err = c.load(ctx, "r", "a", nil) | |
| 207 | + | assert.ErrorIs(t, err, context.DeadlineExceeded, "it joins the build and leaves with its ctx") | |
| 208 | + | ||
| 209 | + | close(release) | |
| 210 | + | require.NoError(t, <-first) | |
| 211 | + | require.NoError(t, <-second) | |
| 212 | + | assert.True(t, secondBuilt.Load()) | |
| 213 | + | assert.True(t, c.has("a") && c.has("b")) | |
| 214 | + | assert.Empty(t, c.building) | |
| 215 | + | assert.Empty(t, c.slots, "every slot is back") | |
| 216 | + | } | |
| 217 | + | ||
| 218 | + | // The slot of a build is kept after its callers leave, and given back when it | |
| 219 | + | // ends, whether it fails or panics. | |
| 220 | + | func TestPathCacheSlotOutlivesCallers(t *testing.T) { | |
| 221 | + | c := newPathCache(1000, 1) | |
| 222 | + | started, release := make(chan struct{}), make(chan struct{}) | |
| 223 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 224 | + | done := make(chan error, 1) | |
| 225 | + | go func() { | |
| 226 | + | _, err := c.load(ctx, "r", "a", func() ([]byte, error) { close(started); <-release; panic("boom") }) | |
| 227 | + | done <- err | |
| 228 | + | }() | |
| 229 | + | <-started | |
| 230 | + | cancel() | |
| 231 | + | assert.ErrorIs(t, <-done, context.Canceled) | |
| 232 | + | assert.Len(t, c.slots, 1, "the build still holds its slot") | |
| 233 | + | close(release) | |
| 234 | + | require.Eventually(t, func() bool { return len(c.slots) == 0 && building(c, 0) }, time.Second, time.Millisecond) | |
| 235 | + | b, err := c.load(context.Background(), "r", "b", func() ([]byte, error) { return []byte("ok"), nil }) | |
| 236 | + | require.NoError(t, err) | |
| 237 | + | assert.Equal(t, "ok", string(b)) | |
| 238 | + | } | |
| 239 | + | ||
| 129 | 240 | // A build that panics is handed to its waiters as an error, leaves nothing | |
| 130 | 241 | // in flight or cached, and does not take the process down. | |
| 131 | 242 | func TestPathCacheLoadPanic(t *testing.T) { | |
| 132 | - | c := newPathCache(1000, 200) | |
| 243 | + | c := newPathCache(1000, 4) | |
| 133 | 244 | started, release := make(chan struct{}), make(chan struct{}) | |
| 134 | 245 | var once sync.Once | |
| 135 | 246 | build := func() ([]byte, error) { | |
| @@ -151,8 +262,7 @@ func TestPathCacheLoadPanic(t *testing.T) { | |||
| 151 | 262 | assert.ErrorIs(t, <-errs, errBuildPanicked) | |
| 152 | 263 | } | |
| 153 | 264 | assert.Empty(t, c.building) | |
| 154 | - | _, ok := c.get("x") | |
| 155 | - | assert.False(t, ok, "a panic is not cached") | |
| 265 | + | assert.False(t, c.has("x"), "a panic is not cached") | |
| 156 | 266 | ||
| 157 | 267 | b, err := c.load(context.Background(), "r", "x", func() ([]byte, error) { return []byte("ok"), nil }) | |
| 158 | 268 | require.NoError(t, err) | |
| @@ -162,7 +272,7 @@ func TestPathCacheLoadPanic(t *testing.T) { | |||
| 162 | 272 | // The caller that starts a build is released with its context too, and the | |
| 163 | 273 | // build goes on for the others. | |
| 164 | 274 | func TestPathCacheLoadStarterLeaves(t *testing.T) { | |
| 165 | - | c := newPathCache(1000, 200) | |
| 275 | + | c := newPathCache(1000, 4) | |
| 166 | 276 | release := make(chan struct{}) | |
| 167 | 277 | build := func() ([]byte, error) { <-release; return []byte("list"), nil } | |
| 168 | 278 | ctx, cancel := context.WithCancel(context.Background()) | |
| @@ -177,14 +287,14 @@ func TestPathCacheLoadStarterLeaves(t *testing.T) { | |||
| 177 | 287 | t.Fatal("the starter is held by its build") | |
| 178 | 288 | } | |
| 179 | 289 | close(release) | |
| 180 | - | require.Eventually(t, func() bool { _, ok := c.get("x"); return ok }, time.Second, time.Millisecond) | |
| 290 | + | require.Eventually(t, func() bool { return c.has("x") }, time.Second, time.Millisecond) | |
| 181 | 291 | assert.Empty(t, c.building) | |
| 182 | 292 | } | |
| 183 | 293 | ||
| 184 | 294 | // A failure belongs to the repository that built it: another repository | |
| 185 | 295 | // asking for the same commit builds its own. | |
| 186 | 296 | func TestPathCacheLoadFailureStaysInRepo(t *testing.T) { | |
| 187 | - | c := newPathCache(1000, 200) | |
| 297 | + | c := newPathCache(1000, 4) | |
| 188 | 298 | started, release := make(chan struct{}), make(chan struct{}) | |
| 189 | 299 | boom := errors.New("boom") | |
| 190 | 300 | failing := make(chan error, 1) | |
| @@ -15,55 +15,80 @@ import ( | |||||
| 15 | "github.com/stretchr/testify/require" | 15 | "github.com/stretchr/testify/require" | ||
| 16 | ) | 16 | ) | ||
| 17 | 17 | ||||
| 18 | + | // store puts a list in the cache the way a finished build does. | |||
| 19 | + | func (c *pathCache) store(id string, body []byte) { | |||
| 20 | + | c.mu.Lock() | |||
| 21 | + | defer c.mu.Unlock() | |||
| 22 | + | c.putLocked(id, body) | |||
| 23 | + | } | |||
| 24 | + | ||||
| 25 | + | // has says whether a list is cached, marking it as used. | |||
| 26 | + | func (c *pathCache) has(id string) bool { | |||
| 27 | + | c.mu.Lock() | |||
| 28 | + | defer c.mu.Unlock() | |||
| 29 | + | _, ok := c.getLocked(id) | |||
| 30 | + | return ok | |||
| 31 | + | } | |||
| 32 | + | ||||
| 18 | func TestPathCacheEvictsLeastRecentlyUsed(t *testing.T) { | 33 | func TestPathCacheEvictsLeastRecentlyUsed(t *testing.T) { | ||
| 19 | - | c := newPathCache(100, 60) | 34 | + | c := newPathCache(100, 1) |
| 20 | blob := func(n int) []byte { return make([]byte, n) } | 35 | blob := func(n int) []byte { return make([]byte, n) } | ||
| 21 | - | c.put("a", blob(40)) | 36 | + | c.store("a", blob(25)) |
| 22 | - | c.put("b", blob(40)) | 37 | + | c.store("b", blob(25)) |
| 23 | - | _, ok := c.get("a") // a is now the most recent | 38 | + | c.store("e", blob(25)) |
| 24 | - | require.True(t, ok) | 39 | + | require.True(t, c.has("a")) // a is now the most recent |
| 25 | - | c.put("c", blob(40)) // 120 > 100: b, the least recent, goes | 40 | + | c.store("c", blob(25)) // 100 fit |
| 26 | - | _, ok = c.get("b") | 41 | + | c.store("d", blob(10)) // 110 > 100: b, the least recent, goes |
| 27 | - | assert.False(t, ok, "b was evicted") | 42 | + | assert.False(t, c.has("b"), "b was evicted") |
| 28 | - | _, okA := c.get("a") | 43 | + | assert.True(t, c.has("a") && c.has("c") && c.has("d") && c.has("e")) |
| 29 | - | _, okC := c.get("c") | 44 | + | assert.Equal(t, 85, c.size) |
| 30 | - | assert.True(t, okA && okC) | |||
| 31 | - | assert.Equal(t, 80, c.size) | |||
| 32 | 45 | ||||
| 33 | // Replacing an entry accounts for the old size. | 46 | // Replacing an entry accounts for the old size. | ||
| 34 | - | c.put("a", blob(10)) | 47 | + | c.store("a", blob(10)) |
| 35 | - | assert.Equal(t, 50, c.size) | 48 | + | assert.Equal(t, 70, c.size) |
| 36 | 49 | ||||
| 37 | // A big one pushes out as many as needed. | 50 | // A big one pushes out as many as needed. | ||
| 38 | - | c.put("d", blob(60)) | 51 | + | c.store("f", blob(25)) |
| 52 | + | c.store("g", blob(25)) | |||
| 39 | assert.LessOrEqual(t, c.size, 100) | 53 | assert.LessOrEqual(t, c.size, 100) | ||
| 40 | - | _, ok = c.get("d") | 54 | + | assert.True(t, c.has("g")) |
| 41 | - | assert.True(t, ok) | |||
| 42 | } | 55 | } | ||
| 43 | 56 | ||||
| 57 | + | // A list over a quarter of the budget is not kept, since it would push out the | |||
| 58 | + | // others; one at the limit is. | |||
| 44 | func TestPathCacheSkipsOversizedEntries(t *testing.T) { | 59 | func TestPathCacheSkipsOversizedEntries(t *testing.T) { | ||
| 45 | - | c := newPathCache(100, 60) | 60 | + | c := newPathCache(100, 1) |
| 46 | - | c.put("big", make([]byte, 61)) | 61 | + | var builds atomic.Int32 |
| 47 | - | _, ok := c.get("big") | 62 | + | big := func(n int) func() ([]byte, error) { |
| 48 | - | assert.False(t, ok, "over the entry limit") | 63 | + | return func() ([]byte, error) { builds.Add(1); return make([]byte, n), nil } |
| 49 | - | c = newPathCache(50, 60) | 64 | + | } |
| 50 | - | c.put("big", make([]byte, 51)) | 65 | + | for range 2 { |
| 51 | - | _, ok = c.get("big") | 66 | + | b, err := c.load(context.Background(), "r", "big", big(26)) |
| 52 | - | assert.False(t, ok, "over the budget") | 67 | + | require.NoError(t, err) |
| 68 | + | assert.Len(t, b, 26, "the caller still gets the list") | |||
| 69 | + | } | |||
| 70 | + | assert.EqualValues(t, 2, builds.Load(), "not cached, so built again") | |||
| 71 | + | assert.False(t, c.has("big")) | |||
| 53 | assert.Zero(t, c.size) | 72 | assert.Zero(t, c.size) | ||
| 73 | + | ||||
| 74 | + | for range 2 { | |||
| 75 | + | _, err := c.load(context.Background(), "r", "fits", big(25)) | |||
| 76 | + | require.NoError(t, err) | |||
| 77 | + | } | |||
| 78 | + | assert.EqualValues(t, 3, builds.Load(), "at a quarter: kept") | |||
| 79 | + | assert.Equal(t, 25, c.size) | |||
| 54 | } | 80 | } | ||
| 55 | 81 | ||||
| 56 | func TestPathCacheConcurrent(t *testing.T) { | 82 | func TestPathCacheConcurrent(t *testing.T) { | ||
| 57 | - | c := newPathCache(1000, 200) | 83 | + | c := newPathCache(1000, 1) |
| 58 | var wg sync.WaitGroup | 84 | var wg sync.WaitGroup | ||
| 59 | for g := range 8 { | 85 | for g := range 8 { | ||
| 60 | wg.Add(1) | 86 | wg.Add(1) | ||
| 61 | go func() { | 87 | go func() { | ||
| 62 | defer wg.Done() | 88 | defer wg.Done() | ||
| 63 | for i := range 500 { | 89 | for i := range 500 { | ||
| 64 | - | id := fmt.Sprint((g + i) % 17) | 90 | + | c.store(fmt.Sprint((g+i)%17), make([]byte, 10+i%90)) |
| 65 | - | c.put(id, make([]byte, 10+i%90)) | 91 | + | c.has(fmt.Sprint(i % 17)) |
| 66 | - | c.get(fmt.Sprint(i % 17)) | |||
| 67 | } | 92 | } | ||
| 68 | }() | 93 | }() | ||
| 69 | } | 94 | } | ||
| @@ -77,10 +102,27 @@ func TestPathCacheConcurrent(t *testing.T) { | |||||
| 77 | assert.Equal(t, len(c.byID), c.lru.Len()) | 102 | assert.Equal(t, len(c.byID), c.lru.Len()) | ||
| 78 | } | 103 | } | ||
| 79 | 104 | ||||
| 105 | + | // waiters is the number of callers waiting for the build of a key, -1 when | |||
| 106 | + | // none is in flight. | |||
| 107 | + | func waiters(c *pathCache, repo, id string) int { | |||
| 108 | + | c.mu.Lock() | |||
| 109 | + | defer c.mu.Unlock() | |||
| 110 | + | if f := c.building[repo+"\x00"+id]; f != nil { | |||
| 111 | + | return f.waiters | |||
| 112 | + | } | |||
| 113 | + | return -1 | |||
| 114 | + | } | |||
| 115 | + | ||||
| 116 | + | func building(c *pathCache, n int) bool { | |||
| 117 | + | c.mu.Lock() | |||
| 118 | + | defer c.mu.Unlock() | |||
| 119 | + | return len(c.building) == n | |||
| 120 | + | } | |||
| 121 | + | ||||
| 80 | // Concurrent misses for one commit build once; a failure is shared and not | 122 | // Concurrent misses for one commit build once; a failure is shared and not | ||
| 81 | // stored; a waiter that gives up leaves the build running. | 123 | // stored; a waiter that gives up leaves the build running. | ||
| 82 | func TestPathCacheLoadBuildsOnce(t *testing.T) { | 124 | func TestPathCacheLoadBuildsOnce(t *testing.T) { | ||
| 83 | - | c := newPathCache(1000, 200) | 125 | + | c := newPathCache(1000, 4) |
| 84 | var builds atomic.Int32 | 126 | var builds atomic.Int32 | ||
| 85 | release := make(chan struct{}) | 127 | release := make(chan struct{}) | ||
| 86 | build := func() ([]byte, error) { | 128 | build := func() ([]byte, error) { | ||
| @@ -88,8 +130,9 @@ func TestPathCacheLoadBuildsOnce(t *testing.T) { | |||||
| 88 | <-release | 130 | <-release | ||
| 89 | return []byte("list"), nil | 131 | return []byte("list"), nil | ||
| 90 | } | 132 | } | ||
| 133 | + | const callers = 10 | |||
| 91 | var wg sync.WaitGroup | 134 | var wg sync.WaitGroup | ||
| 92 | - | got := make([][]byte, 10) | 135 | + | got := make([][]byte, callers) |
| 93 | for i := range got { | 136 | for i := range got { | ||
| 94 | wg.Add(1) | 137 | wg.Add(1) | ||
| 95 | go func() { | 138 | go func() { | ||
| @@ -99,8 +142,10 @@ func TestPathCacheLoadBuildsOnce(t *testing.T) { | |||||
| 99 | got[i] = b | 142 | got[i] = b | ||
| 100 | }() | 143 | }() | ||
| 101 | } | 144 | } | ||
| 145 | + | // Every caller waits for the one build before it is released: none can | |||
| 146 | + | // find the list cached, so one build serving all ten shows they joined it. | |||
| 147 | + | require.Eventually(t, func() bool { return waiters(c, "r", "x") == callers }, 5*time.Second, time.Millisecond) | |||
| 102 | // A waiter whose context ends does not stop the others. | 148 | // A waiter whose context ends does not stop the others. | ||
| 103 | - | require.Eventually(t, func() bool { c.mu.Lock(); defer c.mu.Unlock(); return c.building["r\x00x"] != nil }, time.Second, time.Millisecond) | |||
| 104 | ctx, cancel := context.WithCancel(context.Background()) | 149 | ctx, cancel := context.WithCancel(context.Background()) | ||
| 105 | cancel() | 150 | cancel() | ||
| 106 | _, err := c.load(ctx, "r", "x", build) | 151 | _, err := c.load(ctx, "r", "x", build) | ||
| @@ -111,25 +156,91 @@ func TestPathCacheLoadBuildsOnce(t *testing.T) { | |||||
| 111 | for _, b := range got { | 156 | for _, b := range got { | ||
| 112 | assert.Equal(t, "list", string(b)) | 157 | assert.Equal(t, "list", string(b)) | ||
| 113 | } | 158 | } | ||
| 114 | - | b, ok := c.get("x") | 159 | + | assert.True(t, c.has("x")) |
| 115 | - | assert.True(t, ok) | |||
| 116 | - | assert.Equal(t, "list", string(b)) | |||
| 117 | 160 | ||||
| 118 | boom := errors.New("boom") | 161 | boom := errors.New("boom") | ||
| 119 | _, err = c.load(context.Background(), "r", "y", func() ([]byte, error) { return nil, boom }) | 162 | _, err = c.load(context.Background(), "r", "y", func() ([]byte, error) { return nil, boom }) | ||
| 120 | assert.ErrorIs(t, err, boom) | 163 | assert.ErrorIs(t, err, boom) | ||
| 121 | - | _, ok = c.get("y") | 164 | + | assert.False(t, c.has("y"), "a failure is not cached") |
| 122 | - | assert.False(t, ok, "a failure is not cached") | 165 | + | b, err := c.load(context.Background(), "r", "y", func() ([]byte, error) { return []byte("ok"), nil }) |
| 123 | - | b, err = c.load(context.Background(), "r", "y", func() ([]byte, error) { return []byte("ok"), nil }) | |||
| 124 | require.NoError(t, err) | 166 | require.NoError(t, err) | ||
| 125 | assert.Equal(t, "ok", string(b), "the next request builds again") | 167 | assert.Equal(t, "ok", string(b), "the next request builds again") | ||
| 126 | assert.Empty(t, c.building) | 168 | assert.Empty(t, c.building) | ||
| 127 | } | 169 | } | ||
| 128 | 170 | ||||
| 171 | + | // The builds of all commits share the slots: with one slot, a load of another | |||
| 172 | + | // commit waits for the running build, and leaves with its ctx if it is | |||
| 173 | + | // cancelled first. The slot is the build's, not the caller's. | |||
| 174 | + | func TestPathCacheLoadBoundsBuilds(t *testing.T) { | |||
| 175 | + | c := newPathCache(1000, 1) | |||
| 176 | + | started, release := make(chan struct{}), make(chan struct{}) | |||
| 177 | + | first := make(chan error, 1) | |||
| 178 | + | go func() { | |||
| 179 | + | _, err := c.load(context.Background(), "r", "a", func() ([]byte, error) { close(started); <-release; return []byte("a"), nil }) | |||
| 180 | + | first <- err | |||
| 181 | + | }() | |||
| 182 | + | <-started | |||
| 183 | + | ||||
| 184 | + | // A second commit waits for the slot... | |||
| 185 | + | var secondBuilt atomic.Bool | |||
| 186 | + | second := make(chan error, 1) | |||
| 187 | + | go func() { | |||
| 188 | + | _, err := c.load(context.Background(), "r", "b", func() ([]byte, error) { secondBuilt.Store(true); return []byte("b"), nil }) | |||
| 189 | + | second <- err | |||
| 190 | + | }() | |||
| 191 | + | time.Sleep(30 * time.Millisecond) | |||
| 192 | + | assert.False(t, secondBuilt.Load(), "no slot, no build") | |||
| 193 | + | select { | |||
| 194 | + | case <-second: | |||
| 195 | + | t.Fatal("the second load did not wait for a slot") | |||
| 196 | + | default: | |||
| 197 | + | } | |||
| 198 | + | ||||
| 199 | + | // ...or leaves with its ctx. | |||
| 200 | + | ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) | |||
| 201 | + | defer cancel() | |||
| 202 | + | _, err := c.load(ctx, "r", "c", func() ([]byte, error) { t.Error("built without a slot"); return nil, nil }) | |||
| 203 | + | assert.ErrorIs(t, err, context.DeadlineExceeded) | |||
| 204 | + | ||||
| 205 | + | // A caller of the running build is not held by the slots. | |||
| 206 | + | _, err = c.load(ctx, "r", "a", nil) | |||
| 207 | + | assert.ErrorIs(t, err, context.DeadlineExceeded, "it joins the build and leaves with its ctx") | |||
| 208 | + | ||||
| 209 | + | close(release) | |||
| 210 | + | require.NoError(t, <-first) | |||
| 211 | + | require.NoError(t, <-second) | |||
| 212 | + | assert.True(t, secondBuilt.Load()) | |||
| 213 | + | assert.True(t, c.has("a") && c.has("b")) | |||
| 214 | + | assert.Empty(t, c.building) | |||
| 215 | + | assert.Empty(t, c.slots, "every slot is back") | |||
| 216 | + | } | |||
| 217 | + | ||||
| 218 | + | // The slot of a build is kept after its callers leave, and given back when it | |||
| 219 | + | // ends, whether it fails or panics. | |||
| 220 | + | func TestPathCacheSlotOutlivesCallers(t *testing.T) { | |||
| 221 | + | c := newPathCache(1000, 1) | |||
| 222 | + | started, release := make(chan struct{}), make(chan struct{}) | |||
| 223 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 224 | + | done := make(chan error, 1) | |||
| 225 | + | go func() { | |||
| 226 | + | _, err := c.load(ctx, "r", "a", func() ([]byte, error) { close(started); <-release; panic("boom") }) | |||
| 227 | + | done <- err | |||
| 228 | + | }() | |||
| 229 | + | <-started | |||
| 230 | + | cancel() | |||
| 231 | + | assert.ErrorIs(t, <-done, context.Canceled) | |||
| 232 | + | assert.Len(t, c.slots, 1, "the build still holds its slot") | |||
| 233 | + | close(release) | |||
| 234 | + | require.Eventually(t, func() bool { return len(c.slots) == 0 && building(c, 0) }, time.Second, time.Millisecond) | |||
| 235 | + | b, err := c.load(context.Background(), "r", "b", func() ([]byte, error) { return []byte("ok"), nil }) | |||
| 236 | + | require.NoError(t, err) | |||
| 237 | + | assert.Equal(t, "ok", string(b)) | |||
| 238 | + | } | |||
| 239 | + | ||||
| 129 | // A build that panics is handed to its waiters as an error, leaves nothing | 240 | // A build that panics is handed to its waiters as an error, leaves nothing | ||
| 130 | // in flight or cached, and does not take the process down. | 241 | // in flight or cached, and does not take the process down. | ||
| 131 | func TestPathCacheLoadPanic(t *testing.T) { | 242 | func TestPathCacheLoadPanic(t *testing.T) { | ||
| 132 | - | c := newPathCache(1000, 200) | 243 | + | c := newPathCache(1000, 4) |
| 133 | started, release := make(chan struct{}), make(chan struct{}) | 244 | started, release := make(chan struct{}), make(chan struct{}) | ||
| 134 | var once sync.Once | 245 | var once sync.Once | ||
| 135 | build := func() ([]byte, error) { | 246 | build := func() ([]byte, error) { | ||
| @@ -151,8 +262,7 @@ func TestPathCacheLoadPanic(t *testing.T) { | |||||
| 151 | assert.ErrorIs(t, <-errs, errBuildPanicked) | 262 | assert.ErrorIs(t, <-errs, errBuildPanicked) | ||
| 152 | } | 263 | } | ||
| 153 | assert.Empty(t, c.building) | 264 | assert.Empty(t, c.building) | ||
| 154 | - | _, ok := c.get("x") | 265 | + | assert.False(t, c.has("x"), "a panic is not cached") |
| 155 | - | assert.False(t, ok, "a panic is not cached") | |||
| 156 | 266 | ||||
| 157 | b, err := c.load(context.Background(), "r", "x", func() ([]byte, error) { return []byte("ok"), nil }) | 267 | b, err := c.load(context.Background(), "r", "x", func() ([]byte, error) { return []byte("ok"), nil }) | ||
| 158 | require.NoError(t, err) | 268 | require.NoError(t, err) | ||
| @@ -162,7 +272,7 @@ func TestPathCacheLoadPanic(t *testing.T) { | |||||
| 162 | // The caller that starts a build is released with its context too, and the | 272 | // The caller that starts a build is released with its context too, and the | ||
| 163 | // build goes on for the others. | 273 | // build goes on for the others. | ||
| 164 | func TestPathCacheLoadStarterLeaves(t *testing.T) { | 274 | func TestPathCacheLoadStarterLeaves(t *testing.T) { | ||
| 165 | - | c := newPathCache(1000, 200) | 275 | + | c := newPathCache(1000, 4) |
| 166 | release := make(chan struct{}) | 276 | release := make(chan struct{}) | ||
| 167 | build := func() ([]byte, error) { <-release; return []byte("list"), nil } | 277 | build := func() ([]byte, error) { <-release; return []byte("list"), nil } | ||
| 168 | ctx, cancel := context.WithCancel(context.Background()) | 278 | ctx, cancel := context.WithCancel(context.Background()) | ||
| @@ -177,14 +287,14 @@ func TestPathCacheLoadStarterLeaves(t *testing.T) { | |||||
| 177 | t.Fatal("the starter is held by its build") | 287 | t.Fatal("the starter is held by its build") | ||
| 178 | } | 288 | } | ||
| 179 | close(release) | 289 | close(release) | ||
| 180 | - | require.Eventually(t, func() bool { _, ok := c.get("x"); return ok }, time.Second, time.Millisecond) | 290 | + | require.Eventually(t, func() bool { return c.has("x") }, time.Second, time.Millisecond) |
| 181 | assert.Empty(t, c.building) | 291 | assert.Empty(t, c.building) | ||
| 182 | } | 292 | } | ||
| 183 | 293 | ||||
| 184 | // A failure belongs to the repository that built it: another repository | 294 | // A failure belongs to the repository that built it: another repository | ||
| 185 | // asking for the same commit builds its own. | 295 | // asking for the same commit builds its own. | ||
| 186 | func TestPathCacheLoadFailureStaysInRepo(t *testing.T) { | 296 | func TestPathCacheLoadFailureStaysInRepo(t *testing.T) { | ||
| 187 | - | c := newPathCache(1000, 200) | 297 | + | c := newPathCache(1000, 4) |
| 188 | started, release := make(chan struct{}), make(chan struct{}) | 298 | started, release := make(chan struct{}), make(chan struct{}) | ||
| 189 | boom := errors.New("boom") | 299 | boom := errors.New("boom") | ||
| 190 | failing := make(chan error, 1) | 300 | failing := make(chan error, 1) | ||
M internal/adapters/web/web.go +1 -1
| @@ -111,7 +111,7 @@ func New(cfg Config) (*Server, error) { | |||
| 111 | 111 | } | |
| 112 | 112 | s := &Server{ | |
| 113 | 113 | cfg: cfg, instance: u.Host, assets: assets, md: newMarkdown(), | |
| 114 | - | pathCache: newPathCache(pathCacheBytes, pathEntryMax), | |
| 114 | + | pathCache: newPathCache(pathCacheBytes, pathBuilds), | |
| 115 | 115 | origin: originOf(u), | |
| 116 | 116 | } | |
| 117 | 117 | s.mux = s.routes() | |
| @@ -111,7 +111,7 @@ func New(cfg Config) (*Server, error) { | |||||
| 111 | } | 111 | } | ||
| 112 | s := &Server{ | 112 | s := &Server{ | ||
| 113 | cfg: cfg, instance: u.Host, assets: assets, md: newMarkdown(), | 113 | cfg: cfg, instance: u.Host, assets: assets, md: newMarkdown(), | ||
| 114 | - | pathCache: newPathCache(pathCacheBytes, pathEntryMax), | 114 | + | pathCache: newPathCache(pathCacheBytes, pathBuilds), |
| 115 | origin: originOf(u), | 115 | origin: originOf(u), | ||
| 116 | } | 116 | } | ||
| 117 | s.mux = s.routes() | 117 | s.mux = s.routes() | ||