1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
11 "git.arvados.org/arvados.git/lib/dispatchcloud/container"
12 "git.arvados.org/arvados.git/sdk/go/arvados"
13 "github.com/sirupsen/logrus"
16 var quietAfter503 = time.Minute
18 func (sch *Scheduler) runQueue() {
19 running := sch.pool.Running()
20 unalloc := sch.pool.Unallocated()
22 unsorted, _ := sch.queue.Entries()
23 sorted := make([]container.QueueEnt, 0, len(unsorted))
24 for _, ent := range unsorted {
25 sorted = append(sorted, ent)
27 sort.Slice(sorted, func(i, j int) bool {
28 _, irunning := running[sorted[i].Container.UUID]
29 _, jrunning := running[sorted[j].Container.UUID]
30 if irunning != jrunning {
31 // Ensure the "tryrun" loop (see below) sees
32 // already-scheduled containers first, to
33 // ensure existing supervisor containers are
34 // properly counted before we decide whether
35 // we have room for new ones.
38 ilocked := sorted[i].Container.State == arvados.ContainerStateLocked
39 jlocked := sorted[j].Container.State == arvados.ContainerStateLocked
40 if ilocked != jlocked {
41 // Give precedence to containers that we have
42 // already locked, even if higher-priority
43 // containers have since arrived in the
44 // queue. This avoids undesirable queue churn
45 // effects including extra lock/unlock cycles
46 // and bringing up new instances and quickly
47 // shutting them down to make room for
48 // different instance sizes.
50 } else if pi, pj := sorted[i].Container.Priority, sorted[j].Container.Priority; pi != pj {
53 // When containers have identical priority,
54 // start them in the order we first noticed
55 // them. This avoids extra lock/unlock cycles
56 // when we unlock the containers that don't
57 // fit in the available pool.
58 return sorted[i].FirstSeenAt.Before(sorted[j].FirstSeenAt)
62 if t := sch.client.Last503(); t.After(sch.last503time) {
63 // API has sent an HTTP 503 response since last time
64 // we checked. Use current #containers - 1 as
65 // maxConcurrency, i.e., try to stay just below the
66 // level where we see 503s.
68 if newlimit := len(running) - 1; newlimit < 1 {
69 sch.maxConcurrency = 1
71 sch.maxConcurrency = newlimit
73 } else if sch.maxConcurrency > 0 && time.Since(sch.last503time) > quietAfter503 {
74 // If we haven't seen any 503 errors lately, raise
75 // limit to ~10% beyond the current workload.
77 // As we use the added 10% to schedule more
78 // containers, len(running) will increase and we'll
79 // push the limit up further. Soon enough,
80 // maxConcurrency will get high enough to schedule the
81 // entire queue, hit pool quota, or get 503s again.
82 max := len(running)*11/10 + 1
83 if sch.maxConcurrency < max {
84 sch.maxConcurrency = max
87 if sch.last503time.IsZero() {
88 sch.mLast503Time.Set(0)
90 sch.mLast503Time.Set(float64(sch.last503time.Unix()))
92 if sch.maxInstances > 0 && sch.maxConcurrency > sch.maxInstances {
93 sch.maxConcurrency = sch.maxInstances
95 if sch.pool.AtQuota() && len(running) > 0 && (sch.maxConcurrency == 0 || sch.maxConcurrency > len(running)) {
96 // Consider current workload to be the maximum
97 // allowed, for the sake of reporting metrics and
98 // calculating max supervisors.
100 // Now that sch.maxConcurrency is set, we will only
101 // raise it past len(running) by 10%. This helps
102 // avoid running an inappropriate number of
103 // supervisors when we reach the cloud-imposed quota
104 // (which may be based on # CPUs etc) long before the
105 // configured MaxInstances.
106 sch.maxConcurrency = len(running)
108 sch.mMaxContainerConcurrency.Set(float64(sch.maxConcurrency))
110 maxSupervisors := int(float64(sch.maxConcurrency) * sch.supervisorFraction)
111 if maxSupervisors < 1 && sch.supervisorFraction > 0 && sch.maxConcurrency > 0 {
115 sch.logger.WithFields(logrus.Fields{
116 "Containers": len(sorted),
117 "Processes": len(running),
118 "maxConcurrency": sch.maxConcurrency,
121 dontstart := map[arvados.InstanceType]bool{}
122 var overquota []container.QueueEnt // entries that are unmappable because of worker pool quota
123 var overmaxsuper []container.QueueEnt // unmappable because max supervisors (these are not included in overquota)
124 var containerAllocatedWorkerBootingCount int
126 // trying is #containers running + #containers we're trying to
127 // start. We stop trying to start more containers if this
128 // reaches the dynamic maxConcurrency limit.
129 trying := len(running)
134 for i, ent := range sorted {
135 ctr, it := ent.Container, ent.InstanceType
136 logger := sch.logger.WithFields(logrus.Fields{
137 "ContainerUUID": ctr.UUID,
138 "InstanceType": it.Name,
140 if ctr.SchedulingParameters.Supervisor {
142 if maxSupervisors > 0 && supervisors > maxSupervisors {
143 overmaxsuper = append(overmaxsuper, sorted[i])
147 if _, running := running[ctr.UUID]; running || ctr.Priority < 1 {
151 case arvados.ContainerStateQueued:
152 if sch.maxConcurrency > 0 && trying >= sch.maxConcurrency {
153 logger.Tracef("not locking: already at maxConcurrency %d", sch.maxConcurrency)
157 if unalloc[it] < 1 && sch.pool.AtQuota() {
158 logger.Trace("not locking: AtQuota and no unalloc workers")
159 overquota = sorted[i:]
162 if sch.pool.KillContainer(ctr.UUID, "about to lock") {
163 logger.Info("not locking: crunch-run process from previous attempt has not exited")
166 go sch.lockContainer(logger, ctr.UUID)
168 case arvados.ContainerStateLocked:
169 if sch.maxConcurrency > 0 && trying >= sch.maxConcurrency {
170 logger.Tracef("not starting: already at maxConcurrency %d", sch.maxConcurrency)
176 } else if sch.pool.AtQuota() {
177 // Don't let lower-priority containers
178 // starve this one by using keeping
179 // idle workers alive on different
181 logger.Trace("overquota")
182 overquota = sorted[i:]
184 } else if sch.pool.Create(it) {
185 // Success. (Note pool.Create works
186 // asynchronously and does its own
187 // logging about the eventual outcome,
188 // so we don't need to.)
189 logger.Info("creating new instance")
191 // Failed despite not being at quota,
192 // e.g., cloud ops throttled. TODO:
193 // avoid getting starved here if
194 // instances of a specific type always
196 logger.Trace("pool declined to create new instance")
201 // We already tried & failed to start
202 // a higher-priority container on the
203 // same instance type. Don't let this
204 // one sneak in ahead of it.
205 } else if sch.pool.KillContainer(ctr.UUID, "about to start") {
206 logger.Info("not restarting yet: crunch-run process from previous attempt has not exited")
207 } else if sch.pool.StartContainer(it, ctr) {
208 logger.Trace("StartContainer => true")
211 logger.Trace("StartContainer => false")
212 containerAllocatedWorkerBootingCount += 1
218 sch.mContainersAllocatedNotStarted.Set(float64(containerAllocatedWorkerBootingCount))
219 sch.mContainersNotAllocatedOverQuota.Set(float64(len(overquota) + len(overmaxsuper)))
221 if len(overquota)+len(overmaxsuper) > 0 {
222 // Unlock any containers that are unmappable while
223 // we're at quota (but if they have already been
224 // scheduled and they're loading docker images etc.,
226 for _, ctr := range append(overmaxsuper, overquota...) {
228 _, toolate := running[ctr.UUID]
229 if ctr.State == arvados.ContainerStateLocked && !toolate {
230 logger := sch.logger.WithField("ContainerUUID", ctr.UUID)
231 logger.Info("unlock because pool capacity is used by higher priority containers")
232 err := sch.queue.Unlock(ctr.UUID)
234 logger.WithError(err).Warn("error unlocking")
239 if len(overquota) > 0 {
240 // Shut down idle workers that didn't get any
241 // containers mapped onto them before we hit quota.
242 for it, n := range unalloc {
246 sch.pool.Shutdown(it)
251 // Lock the given container. Should be called in a new goroutine.
252 func (sch *Scheduler) lockContainer(logger logrus.FieldLogger, uuid string) {
253 if !sch.uuidLock(uuid, "lock") {
256 defer sch.uuidUnlock(uuid)
257 if ctr, ok := sch.queue.Get(uuid); !ok || ctr.State != arvados.ContainerStateQueued {
258 // This happens if the container has been cancelled or
259 // locked since runQueue called sch.queue.Entries(),
260 // possibly by a lockContainer() call from a previous
261 // runQueue iteration. In any case, we will respond
262 // appropriately on the next runQueue iteration, which
263 // will have already been triggered by the queue
265 logger.WithField("State", ctr.State).Debug("container no longer queued by the time we decided to lock it, doing nothing")
268 err := sch.queue.Lock(uuid)
270 logger.WithError(err).Warn("error locking container")
273 logger.Debug("lock succeeded")
274 ctr, ok := sch.queue.Get(uuid)
276 logger.Error("(BUG?) container disappeared from queue after Lock succeeded")
277 } else if ctr.State != arvados.ContainerStateLocked {
278 logger.Warnf("(race?) container has state=%q after Lock succeeded", ctr.State)
282 // Acquire a non-blocking lock for specified UUID, returning true if
283 // successful. The op argument is used only for debug logs.
285 // If the lock is not available, uuidLock arranges to wake up the
286 // scheduler after a short delay, so it can retry whatever operation
287 // is trying to get the lock (if that operation is still worth doing).
289 // This mechanism helps avoid spamming the controller/database with
290 // concurrent updates for any single container, even when the
291 // scheduler loop is running frequently.
292 func (sch *Scheduler) uuidLock(uuid, op string) bool {
294 defer sch.mtx.Unlock()
295 logger := sch.logger.WithFields(logrus.Fields{
296 "ContainerUUID": uuid,
299 if op, locked := sch.uuidOp[uuid]; locked {
300 logger.Debugf("uuidLock not available, Op=%s in progress", op)
301 // Make sure the scheduler loop wakes up to retry.
302 sch.wakeup.Reset(time.Second / 4)
305 logger.Debug("uuidLock acquired")
306 sch.uuidOp[uuid] = op
310 func (sch *Scheduler) uuidUnlock(uuid string) {
312 defer sch.mtx.Unlock()
313 delete(sch.uuidOp, uuid)