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 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)
  21         }
  22         sort.Slice(sorted, func(i, j int) bool {
  23                 return sorted[i].Container.Priority > sorted[j].Container.Priority
  24         })
  25
  26         running := sch.pool.Running()
  27         unalloc := sch.pool.Unallocated()
  28
  29         sch.logger.WithFields(logrus.Fields{
  30                 "Containers": len(sorted),
  31                 "Processes":  len(running),
  32         }).Debug("runQueue")
  33
  34         dontstart := map[arvados.InstanceType]bool{}
  35         var overquota []container.QueueEnt // entries that are unmappable because of worker pool quota
  36
  37 tryrun:
  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,
  43                 })
  44                 if _, running := running[ctr.UUID]; running || ctr.Priority < 1 {
  45                         continue
  46                 }
  47                 switch ctr.State {
  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:]
  52                                 break tryrun
  53                         }
  54                         go sch.lockContainer(logger, ctr.UUID)
  55                         unalloc[it]--
  56                 case arvados.ContainerStateLocked:
  57                         if unalloc[it] > 0 {
  58                                 unalloc[it]--
  59                         } else if sch.pool.AtQuota() {
  60                                 logger.Debug("not starting: AtQuota and no unalloc workers")
  61                                 overquota = sorted[i:]
  62                                 break tryrun
  63                         } else {
  64                                 logger.Info("creating new instance")
  65                                 if !sch.pool.Create(it) {
  66                                         // (Note pool.Create works
  67                                         // asynchronously and logs its
  68                                         // own failures, so we don't
  69                                         // need to log this as a
  70                                         // failure.)
  71
  72                                         sch.queue.Unlock(ctr.UUID)
  73                                         // Don't let lower-priority
  74                                         // containers starve this one
  75                                         // by using keeping idle
  76                                         // workers alive on different
  77                                         // instance types.  TODO:
  78                                         // avoid getting starved here
  79                                         // if instances of a specific
  80                                         // type always fail.
  81                                         overquota = sorted[i:]
  82                                         break tryrun
  83                                 }
  84                         }
  85
  86                         if dontstart[it] {
  87                                 // We already tried & failed to start
  88                                 // a higher-priority container on the
  89                                 // same instance type. Don't let this
  90                                 // one sneak in ahead of it.
  91                         } else if sch.pool.KillContainer(ctr.UUID, "about to lock") {
  92                                 logger.Info("not restarting yet: crunch-run process from previous attempt has not exited")
  93                         } else if sch.pool.StartContainer(it, ctr) {
  94                                 // Success.
  95                         } else {
  96                                 dontstart[it] = true
  97                         }
  98                 }
  99         }
 100
 101         if len(overquota) > 0 {
 102                 // Unlock any containers that are unmappable while
 103                 // we're at quota.
 104                 for _, ctr := range overquota {
 105                         ctr := ctr.Container
 106                         if ctr.State == arvados.ContainerStateLocked {
 107                                 logger := sch.logger.WithField("ContainerUUID", ctr.UUID)
 108                                 logger.Debug("unlock because pool capacity is used by higher priority containers")
 109                                 err := sch.queue.Unlock(ctr.UUID)
 110                                 if err != nil {
 111                                         logger.WithError(err).Warn("error unlocking")
 112                                 }
 113                         }
 114                 }
 115                 // Shut down idle workers that didn't get any
 116                 // containers mapped onto them before we hit quota.
 117                 for it, n := range unalloc {
 118                         if n < 1 {
 119                                 continue
 120                         }
 121                         sch.pool.Shutdown(it)
 122                 }
 123         }
 124 }
 125
 126 // Lock the given container. Should be called in a new goroutine.
 127 func (sch *Scheduler) lockContainer(logger logrus.FieldLogger, uuid string) {
 128         if !sch.uuidLock(uuid, "lock") {
 129                 return
 130         }
 131         defer sch.uuidUnlock(uuid)
 132         if ctr, ok := sch.queue.Get(uuid); !ok || ctr.State != arvados.ContainerStateQueued {
 133                 // This happens if the container has been cancelled or
 134                 // locked since runQueue called sch.queue.Entries(),
 135                 // possibly by a lockContainer() call from a previous
 136                 // runQueue iteration. In any case, we will respond
 137                 // appropriately on the next runQueue iteration, which
 138                 // will have already been triggered by the queue
 139                 // update.
 140                 logger.WithField("State", ctr.State).Debug("container no longer queued by the time we decided to lock it, doing nothing")
 141                 return
 142         }
 143         err := sch.queue.Lock(uuid)
 144         if err != nil {
 145                 logger.WithError(err).Warn("error locking container")
 146                 return
 147         }
 148         logger.Debug("lock succeeded")
 149         ctr, ok := sch.queue.Get(uuid)
 150         if !ok {
 151                 logger.Error("(BUG?) container disappeared from queue after Lock succeeded")
 152         } else if ctr.State != arvados.ContainerStateLocked {
 153                 logger.Warnf("(race?) container has state=%q after Lock succeeded", ctr.State)
 154         }
 155 }
 156
 157 // Acquire a non-blocking lock for specified UUID, returning true if
 158 // successful.  The op argument is used only for debug logs.
 159 //
 160 // If the lock is not available, uuidLock arranges to wake up the
 161 // scheduler after a short delay, so it can retry whatever operation
 162 // is trying to get the lock (if that operation is still worth doing).
 163 //
 164 // This mechanism helps avoid spamming the controller/database with
 165 // concurrent updates for any single container, even when the
 166 // scheduler loop is running frequently.
 167 func (sch *Scheduler) uuidLock(uuid, op string) bool {
 168         sch.mtx.Lock()
 169         defer sch.mtx.Unlock()
 170         logger := sch.logger.WithFields(logrus.Fields{
 171                 "ContainerUUID": uuid,
 172                 "Op":            op,
 173         })
 174         if op, locked := sch.uuidOp[uuid]; locked {
 175                 logger.Debugf("uuidLock not available, Op=%s in progress", op)
 176                 // Make sure the scheduler loop wakes up to retry.
 177                 sch.wakeup.Reset(time.Second / 4)
 178                 return false
 179         }
 180         logger.Debug("uuidLock acquired")
 181         sch.uuidOp[uuid] = op
 182         return true
 183 }
 184
 185 func (sch *Scheduler) uuidUnlock(uuid string) {
 186         sch.mtx.Lock()
 187         defer sch.mtx.Unlock()
 188         delete(sch.uuidOp, uuid)
 189 }