lib/dispatchcloud/scheduler/run_queue.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package scheduler
   6
   7 import (
   8         "sort"
   9         "time"
  10
  11         "git.arvados.org/arvados.git/lib/dispatchcloud/container"
  12         "git.arvados.org/arvados.git/sdk/go/arvados"
  13         "github.com/sirupsen/logrus"
  14 )
  15
  16 var quietAfter503 = time.Minute
  17
  18 func (sch *Scheduler) runQueue() {
  19         unsorted, _ := sch.queue.Entries()
  20         sorted := make([]container.QueueEnt, 0, len(unsorted))
  21         for _, ent := range unsorted {
  22                 sorted = append(sorted, ent)
  23         }
  24         sort.Slice(sorted, func(i, j int) bool {
  25                 if pi, pj := sorted[i].Container.Priority, sorted[j].Container.Priority; pi != pj {
  26                         return pi > pj
  27                 } else {
  28                         // When containers have identical priority,
  29                         // start them in the order we first noticed
  30                         // them. This avoids extra lock/unlock cycles
  31                         // when we unlock the containers that don't
  32                         // fit in the available pool.
  33                         return sorted[i].FirstSeenAt.Before(sorted[j].FirstSeenAt)
  34                 }
  35         })
  36
  37         running := sch.pool.Running()
  38         unalloc := sch.pool.Unallocated()
  39
  40         if t := sch.client.Last503(); t.After(sch.last503time) {
  41                 // API has sent an HTTP 503 response since last time
  42                 // we checked. Use current #containers - 1 as
  43                 // maxConcurrency, i.e., try to stay just below the
  44                 // level where we see 503s.
  45                 sch.last503time = t
  46                 if newlimit := len(running) - 1; newlimit < 1 {
  47                         sch.maxConcurrency = 1
  48                 } else {
  49                         sch.maxConcurrency = newlimit
  50                 }
  51         } else if sch.maxConcurrency > 0 && time.Since(sch.last503time) > quietAfter503 {
  52                 // If we haven't seen any 503 errors lately, raise
  53                 // limit to ~10% beyond the current workload.
  54                 //
  55                 // As we use the added 10% to schedule more
  56                 // containers, len(running) will increase and we'll
  57                 // push the limit up further. Soon enough,
  58                 // maxConcurrency will get high enough to schedule the
  59                 // entire queue, hit pool quota, or get 503s again.
  60                 max := len(running)*11/10 + 1
  61                 if sch.maxConcurrency < max {
  62                         sch.maxConcurrency = max
  63                 }
  64         }
  65         if sch.last503time.IsZero() {
  66                 sch.mLast503Time.Set(0)
  67         } else {
  68                 sch.mLast503Time.Set(float64(sch.last503time.Unix()))
  69         }
  70         sch.mMaxContainerConcurrency.Set(float64(sch.maxConcurrency))
  71
  72         sch.logger.WithFields(logrus.Fields{
  73                 "Containers":     len(sorted),
  74                 "Processes":      len(running),
  75                 "maxConcurrency": sch.maxConcurrency,
  76         }).Debug("runQueue")
  77
  78         dontstart := map[arvados.InstanceType]bool{}
  79         var overquota []container.QueueEnt // entries that are unmappable because of worker pool quota
  80         var containerAllocatedWorkerBootingCount int
  81
  82         // trying is #containers running + #containers we're trying to
  83         // start. We stop trying to start more containers if this
  84         // reaches the dynamic maxConcurrency limit.
  85         trying := len(running)
  86
  87 tryrun:
  88         for i, ctr := range sorted {
  89                 ctr, it := ctr.Container, ctr.InstanceType
  90                 logger := sch.logger.WithFields(logrus.Fields{
  91                         "ContainerUUID": ctr.UUID,
  92                         "InstanceType":  it.Name,
  93                 })
  94                 if _, running := running[ctr.UUID]; running || ctr.Priority < 1 {
  95                         continue
  96                 }
  97                 switch ctr.State {
  98                 case arvados.ContainerStateQueued:
  99                         if sch.maxConcurrency > 0 && trying >= sch.maxConcurrency {
 100                                 logger.Tracef("not locking: already at maxConcurrency %d", sch.maxConcurrency)
 101                                 overquota = sorted[i:]
 102                                 break tryrun
 103                         }
 104                         trying++
 105                         if unalloc[it] < 1 && sch.pool.AtQuota() {
 106                                 logger.Trace("not locking: AtQuota and no unalloc workers")
 107                                 overquota = sorted[i:]
 108                                 break tryrun
 109                         }
 110                         if sch.pool.KillContainer(ctr.UUID, "about to lock") {
 111                                 logger.Info("not locking: crunch-run process from previous attempt has not exited")
 112                                 continue
 113                         }
 114                         go sch.lockContainer(logger, ctr.UUID)
 115                         unalloc[it]--
 116                 case arvados.ContainerStateLocked:
 117                         if sch.maxConcurrency > 0 && trying >= sch.maxConcurrency {
 118                                 logger.Debugf("not starting: already at maxConcurrency %d", sch.maxConcurrency)
 119                                 overquota = sorted[i:]
 120                                 break tryrun
 121                         }
 122                         trying++
 123                         if unalloc[it] > 0 {
 124                                 unalloc[it]--
 125                         } else if sch.pool.AtQuota() {
 126                                 // Don't let lower-priority containers
 127                                 // starve this one by using keeping
 128                                 // idle workers alive on different
 129                                 // instance types.
 130                                 logger.Trace("overquota")
 131                                 overquota = sorted[i:]
 132                                 break tryrun
 133                         } else if sch.pool.Create(it) {
 134                                 // Success. (Note pool.Create works
 135                                 // asynchronously and does its own
 136                                 // logging about the eventual outcome,
 137                                 // so we don't need to.)
 138                                 logger.Info("creating new instance")
 139                         } else {
 140                                 // Failed despite not being at quota,
 141                                 // e.g., cloud ops throttled.  TODO:
 142                                 // avoid getting starved here if
 143                                 // instances of a specific type always
 144                                 // fail.
 145                                 logger.Trace("pool declined to create new instance")
 146                                 continue
 147                         }
 148
 149                         if dontstart[it] {
 150                                 // We already tried & failed to start
 151                                 // a higher-priority container on the
 152                                 // same instance type. Don't let this
 153                                 // one sneak in ahead of it.
 154                         } else if sch.pool.KillContainer(ctr.UUID, "about to start") {
 155                                 logger.Info("not restarting yet: crunch-run process from previous attempt has not exited")
 156                         } else if sch.pool.StartContainer(it, ctr) {
 157                                 // Success.
 158                         } else {
 159                                 containerAllocatedWorkerBootingCount += 1
 160                                 dontstart[it] = true
 161                         }
 162                 }
 163         }
 164
 165         sch.mContainersAllocatedNotStarted.Set(float64(containerAllocatedWorkerBootingCount))
 166         sch.mContainersNotAllocatedOverQuota.Set(float64(len(overquota)))
 167
 168         if len(overquota) > 0 {
 169                 // Unlock any containers that are unmappable while
 170                 // we're at quota (but if they have already been
 171                 // scheduled and they're loading docker images etc.,
 172                 // let them run).
 173                 for _, ctr := range overquota {
 174                         ctr := ctr.Container
 175                         _, toolate := running[ctr.UUID]
 176                         if ctr.State == arvados.ContainerStateLocked && !toolate {
 177                                 logger := sch.logger.WithField("ContainerUUID", ctr.UUID)
 178                                 logger.Debug("unlock because pool capacity is used by higher priority containers")
 179                                 err := sch.queue.Unlock(ctr.UUID)
 180                                 if err != nil {
 181                                         logger.WithError(err).Warn("error unlocking")
 182                                 }
 183                         }
 184                 }
 185                 // Shut down idle workers that didn't get any
 186                 // containers mapped onto them before we hit quota.
 187                 for it, n := range unalloc {
 188                         if n < 1 {
 189                                 continue
 190                         }
 191                         sch.pool.Shutdown(it)
 192                 }
 193         }
 194 }
 195
 196 // Lock the given container. Should be called in a new goroutine.
 197 func (sch *Scheduler) lockContainer(logger logrus.FieldLogger, uuid string) {
 198         if !sch.uuidLock(uuid, "lock") {
 199                 return
 200         }
 201         defer sch.uuidUnlock(uuid)
 202         if ctr, ok := sch.queue.Get(uuid); !ok || ctr.State != arvados.ContainerStateQueued {
 203                 // This happens if the container has been cancelled or
 204                 // locked since runQueue called sch.queue.Entries(),
 205                 // possibly by a lockContainer() call from a previous
 206                 // runQueue iteration. In any case, we will respond
 207                 // appropriately on the next runQueue iteration, which
 208                 // will have already been triggered by the queue
 209                 // update.
 210                 logger.WithField("State", ctr.State).Debug("container no longer queued by the time we decided to lock it, doing nothing")
 211                 return
 212         }
 213         err := sch.queue.Lock(uuid)
 214         if err != nil {
 215                 logger.WithError(err).Warn("error locking container")
 216                 return
 217         }
 218         logger.Debug("lock succeeded")
 219         ctr, ok := sch.queue.Get(uuid)
 220         if !ok {
 221                 logger.Error("(BUG?) container disappeared from queue after Lock succeeded")
 222         } else if ctr.State != arvados.ContainerStateLocked {
 223                 logger.Warnf("(race?) container has state=%q after Lock succeeded", ctr.State)
 224         }
 225 }
 226
 227 // Acquire a non-blocking lock for specified UUID, returning true if
 228 // successful.  The op argument is used only for debug logs.
 229 //
 230 // If the lock is not available, uuidLock arranges to wake up the
 231 // scheduler after a short delay, so it can retry whatever operation
 232 // is trying to get the lock (if that operation is still worth doing).
 233 //
 234 // This mechanism helps avoid spamming the controller/database with
 235 // concurrent updates for any single container, even when the
 236 // scheduler loop is running frequently.
 237 func (sch *Scheduler) uuidLock(uuid, op string) bool {
 238         sch.mtx.Lock()
 239         defer sch.mtx.Unlock()
 240         logger := sch.logger.WithFields(logrus.Fields{
 241                 "ContainerUUID": uuid,
 242                 "Op":            op,
 243         })
 244         if op, locked := sch.uuidOp[uuid]; locked {
 245                 logger.Debugf("uuidLock not available, Op=%s in progress", op)
 246                 // Make sure the scheduler loop wakes up to retry.
 247                 sch.wakeup.Reset(time.Second / 4)
 248                 return false
 249         }
 250         logger.Debug("uuidLock acquired")
 251         sch.uuidOp[uuid] = op
 252         return true
 253 }
 254
 255 func (sch *Scheduler) uuidUnlock(uuid string) {
 256         sch.mtx.Lock()
 257         defer sch.mtx.Unlock()
 258         delete(sch.uuidOp, uuid)
 259 }