Limit concurrent indexing to two runs
Further jobs report a queued phase until a slot frees, so a burst of users cannot exhaust the container's CPU and memory at once.
This commit is contained in:
@@ -45,7 +45,8 @@ type Job struct {
|
||||
|
||||
const (
|
||||
idleExpiry = time.Hour // snapshots unused and jobs untouched for this long are dropped
|
||||
maxSnapshots = 8 // ponytail: fixed cap sized for the 2GB container; make it configurable if memory changes
|
||||
maxSnapshots = 8 // ponytail: fixed caps sized for the 2GB container; make them configurable if memory changes
|
||||
maxIndexing = 2 // concurrent index runs; each already uses every CPU
|
||||
)
|
||||
|
||||
type Server struct {
|
||||
@@ -56,6 +57,7 @@ type Server struct {
|
||||
workspace string // parent of every job directory; only paths under it are ever removed
|
||||
cors string
|
||||
static http.Handler
|
||||
indexSlots chan struct{}
|
||||
}
|
||||
|
||||
func NewServer(workspace string) *Server {
|
||||
@@ -70,6 +72,7 @@ func NewServer(workspace string) *Server {
|
||||
workspace: workspace,
|
||||
cors: cors,
|
||||
static: http.FileServerFS(web),
|
||||
indexSlots: make(chan struct{}, maxIndexing),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -517,6 +520,13 @@ func (s *Server) localJobRoot(id string) (string, bool) {
|
||||
}
|
||||
|
||||
func (s *Server) indexJob(id, root string) {
|
||||
select {
|
||||
case s.indexSlots <- struct{}{}:
|
||||
default:
|
||||
s.updateJob(id, "queued", 0, 0, "Waiting for other indexing to finish…")
|
||||
s.indexSlots <- struct{}{}
|
||||
}
|
||||
defer func() { <-s.indexSlots }()
|
||||
s.updateJob(id, "scanning", 0, 0, "Discovering source files…")
|
||||
snapshot, err := indexDirectory(root, func(phase string, completed, total int, message string) {
|
||||
s.updateJob(id, phase, completed, total, message)
|
||||
|
||||
@@ -361,3 +361,32 @@ func TestSnapshotCapEvictsLeastRecentlyUsed(t *testing.T) {
|
||||
}
|
||||
first.json("GET", "/api/snapshots/"+oldest, "", 404, nil)
|
||||
}
|
||||
|
||||
func TestIndexingQueuesBeyondConcurrencyLimit(t *testing.T) {
|
||||
server, c := newTestServer(t)
|
||||
for range maxIndexing {
|
||||
server.indexSlots <- struct{}{}
|
||||
}
|
||||
var created struct{ JobID string }
|
||||
c.json("POST", "/api/jobs", `{"kind":"local"}`, 201, &created)
|
||||
c.json("POST", "/api/jobs/"+created.JobID+"/files?path=a.go", "package a", 200, nil)
|
||||
c.json("POST", "/api/jobs/"+created.JobID+"/commit", "", 202, nil)
|
||||
var job Job
|
||||
for range 100 {
|
||||
if c.json("GET", "/api/jobs/"+created.JobID, "", 200, &job); job.Phase == "queued" {
|
||||
break
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
if job.Phase != "queued" {
|
||||
t.Fatalf("phase = %q, want queued", job.Phase)
|
||||
}
|
||||
<-server.indexSlots
|
||||
for range 200 {
|
||||
if c.json("GET", "/api/jobs/"+created.JobID, "", 200, &job); job.SnapshotID != nil {
|
||||
return
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("job never finished after a slot was freed: %+v", job)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user