1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
10 "git.curoverse.com/arvados.git/lib/cloud"
11 "git.curoverse.com/arvados.git/lib/dispatchcloud/container"
12 "git.curoverse.com/arvados.git/sdk/go/arvados"
13 "github.com/sirupsen/logrus"
16 func (sch *Scheduler) runQueue() {
17 unsorted, _ := sch.queue.Entries()
18 sorted := make([]container.QueueEnt, 0, len(unsorted))
19 for _, ent := range unsorted {
20 sorted = append(sorted, ent)
22 sort.Slice(sorted, func(i, j int) bool {
23 return sorted[i].Container.Priority > sorted[j].Container.Priority
26 running := sch.pool.Running()
27 unalloc := sch.pool.Unallocated()
29 sch.logger.WithFields(logrus.Fields{
30 "Containers": len(sorted),
31 "Processes": len(running),
34 dontstart := map[arvados.InstanceType]bool{}
35 var overquota []container.QueueEnt // entries that are unmappable because of worker pool quota
38 for i, ctr := range sorted {
39 ctr, it := ctr.Container, ctr.InstanceType
40 logger := sch.logger.WithFields(logrus.Fields{
41 "ContainerUUID": ctr.UUID,
42 "InstanceType": it.Name,
44 if _, running := running[ctr.UUID]; running || ctr.Priority < 1 {
48 case arvados.ContainerStateQueued:
49 if unalloc[it] < 1 && sch.pool.AtQuota() {
50 logger.Debug("not locking: AtQuota and no unalloc workers")
51 overquota = sorted[i:]
54 sch.bgLock(logger, ctr.UUID)
56 case arvados.ContainerStateLocked:
59 } else if sch.pool.AtQuota() {
60 logger.Debug("not starting: AtQuota and no unalloc workers")
61 overquota = sorted[i:]
64 logger.Info("creating new instance")
65 err := sch.pool.Create(it)
67 if _, ok := err.(cloud.QuotaError); !ok {
68 logger.WithError(err).Warn("error creating worker")
70 sch.queue.Unlock(ctr.UUID)
71 // Don't let lower-priority
72 // containers starve this one
73 // by using keeping idle
74 // workers alive on different
75 // instance types. TODO:
76 // avoid getting starved here
77 // if instances of a specific
79 overquota = sorted[i:]
85 // We already tried & failed to start
86 // a higher-priority container on the
87 // same instance type. Don't let this
88 // one sneak in ahead of it.
89 } else if sch.pool.StartContainer(it, ctr) {
97 if len(overquota) > 0 {
98 // Unlock any containers that are unmappable while
100 for _, ctr := range overquota {
102 if ctr.State == arvados.ContainerStateLocked {
103 logger := sch.logger.WithField("ContainerUUID", ctr.UUID)
104 logger.Debug("unlock because pool capacity is used by higher priority containers")
105 err := sch.queue.Unlock(ctr.UUID)
107 logger.WithError(err).Warn("error unlocking")
111 // Shut down idle workers that didn't get any
112 // containers mapped onto them before we hit quota.
113 for it, n := range unalloc {
117 sch.pool.Shutdown(it)
122 // Start an API call to lock the given container, and return
123 // immediately while waiting for the response in a new goroutine. Do
124 // nothing if a lock request is already in progress for this
126 func (sch *Scheduler) bgLock(logger logrus.FieldLogger, uuid string) {
127 logger.Debug("locking")
129 defer sch.mtx.Unlock()
130 if sch.locking[uuid] {
131 logger.Debug("locking in progress, doing nothing")
134 if ctr, ok := sch.queue.Get(uuid); !ok || ctr.State != arvados.ContainerStateQueued {
135 // This happens if the container has been cancelled or
136 // locked since runQueue called sch.queue.Entries(),
137 // possibly by a bgLock() call from a previous
138 // runQueue iteration. In any case, we will respond
139 // appropriately on the next runQueue iteration, which
140 // will have already been triggered by the queue
142 logger.WithField("State", ctr.State).Debug("container no longer queued by the time we decided to lock it, doing nothing")
145 sch.locking[uuid] = true
149 defer sch.mtx.Unlock()
150 delete(sch.locking, uuid)
152 err := sch.queue.Lock(uuid)
154 logger.WithError(err).Warn("error locking container")
157 logger.Debug("lock succeeded")
158 ctr, ok := sch.queue.Get(uuid)
160 logger.Error("(BUG?) container disappeared from queue after Lock succeeded")
161 } else if ctr.State != arvados.ContainerStateLocked {
162 logger.Warnf("(race?) container has state=%q after Lock succeeded", ctr.State)