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
  10         "git.curoverse.com/arvados.git/lib/dispatchcloud/container"
  11         "git.curoverse.com/arvados.git/sdk/go/arvados"
  12         "github.com/sirupsen/logrus"
  13 )
  14
  15 func (sch *Scheduler) runQueue() {
  16         unsorted, _ := sch.queue.Entries()
  17         sorted := make([]container.QueueEnt, 0, len(unsorted))
  18         for _, ent := range unsorted {
  19                 sorted = append(sorted, ent)
  20         }
  21         sort.Slice(sorted, func(i, j int) bool {
  22                 return sorted[i].Container.Priority > sorted[j].Container.Priority
  23         })
  24
  25         running := sch.pool.Running()
  26         unalloc := sch.pool.Unallocated()
  27
  28         sch.logger.WithFields(logrus.Fields{
  29                 "Containers": len(sorted),
  30                 "Processes":  len(running),
  31         }).Debug("runQueue")
  32
  33         dontstart := map[arvados.InstanceType]bool{}
  34         var overquota []container.QueueEnt // entries that are unmappable because of worker pool quota
  35
  36 tryrun:
  37         for i, ctr := range sorted {
  38                 ctr, it := ctr.Container, ctr.InstanceType
  39                 logger := sch.logger.WithFields(logrus.Fields{
  40                         "ContainerUUID": ctr.UUID,
  41                         "InstanceType":  it.Name,
  42                 })
  43                 if _, running := running[ctr.UUID]; running || ctr.Priority < 1 {
  44                         continue
  45                 }
  46                 switch ctr.State {
  47                 case arvados.ContainerStateQueued:
  48                         if unalloc[it] < 1 && sch.pool.AtQuota() {
  49                                 logger.Debug("not locking: AtQuota and no unalloc workers")
  50                                 overquota = sorted[i:]
  51                                 break tryrun
  52                         }
  53                         sch.bgLock(logger, ctr.UUID)
  54                         unalloc[it]--
  55                 case arvados.ContainerStateLocked:
  56                         if unalloc[it] > 0 {
  57                                 unalloc[it]--
  58                         } else if sch.pool.AtQuota() {
  59                                 logger.Debug("not starting: AtQuota and no unalloc workers")
  60                                 overquota = sorted[i:]
  61                                 break tryrun
  62                         } else {
  63                                 logger.Info("creating new instance")
  64                                 if !sch.pool.Create(it) {
  65                                         // (Note pool.Create works
  66                                         // asynchronously and logs its
  67                                         // own failures, so we don't
  68                                         // need to log this as a
  69                                         // failure.)
  70
  71                                         sch.queue.Unlock(ctr.UUID)
  72                                         // Don't let lower-priority
  73                                         // containers starve this one
  74                                         // by using keeping idle
  75                                         // workers alive on different
  76                                         // instance types.  TODO:
  77                                         // avoid getting starved here
  78                                         // if instances of a specific
  79                                         // type always fail.
  80                                         overquota = sorted[i:]
  81                                         break tryrun
  82                                 }
  83                         }
  84
  85                         if dontstart[it] {
  86                                 // We already tried & failed to start
  87                                 // a higher-priority container on the
  88                                 // same instance type. Don't let this
  89                                 // one sneak in ahead of it.
  90                         } else if sch.pool.StartContainer(it, ctr) {
  91                                 // Success.
  92                         } else {
  93                                 dontstart[it] = true
  94                         }
  95                 }
  96         }
  97
  98         if len(overquota) > 0 {
  99                 // Unlock any containers that are unmappable while
 100                 // we're at quota.
 101                 for _, ctr := range overquota {
 102                         ctr := ctr.Container
 103                         if ctr.State == arvados.ContainerStateLocked {
 104                                 logger := sch.logger.WithField("ContainerUUID", ctr.UUID)
 105                                 logger.Debug("unlock because pool capacity is used by higher priority containers")
 106                                 err := sch.queue.Unlock(ctr.UUID)
 107                                 if err != nil {
 108                                         logger.WithError(err).Warn("error unlocking")
 109                                 }
 110                         }
 111                 }
 112                 // Shut down idle workers that didn't get any
 113                 // containers mapped onto them before we hit quota.
 114                 for it, n := range unalloc {
 115                         if n < 1 {
 116                                 continue
 117                         }
 118                         sch.pool.Shutdown(it)
 119                 }
 120         }
 121 }
 122
 123 // Start an API call to lock the given container, and return
 124 // immediately while waiting for the response in a new goroutine. Do
 125 // nothing if a lock request is already in progress for this
 126 // container.
 127 func (sch *Scheduler) bgLock(logger logrus.FieldLogger, uuid string) {
 128         logger.Debug("locking")
 129         sch.mtx.Lock()
 130         defer sch.mtx.Unlock()
 131         if sch.locking[uuid] {
 132                 logger.Debug("locking in progress, doing nothing")
 133                 return
 134         }
 135         if ctr, ok := sch.queue.Get(uuid); !ok || ctr.State != arvados.ContainerStateQueued {
 136                 // This happens if the container has been cancelled or
 137                 // locked since runQueue called sch.queue.Entries(),
 138                 // possibly by a bgLock() call from a previous
 139                 // runQueue iteration. In any case, we will respond
 140                 // appropriately on the next runQueue iteration, which
 141                 // will have already been triggered by the queue
 142                 // update.
 143                 logger.WithField("State", ctr.State).Debug("container no longer queued by the time we decided to lock it, doing nothing")
 144                 return
 145         }
 146         sch.locking[uuid] = true
 147         go func() {
 148                 defer func() {
 149                         sch.mtx.Lock()
 150                         defer sch.mtx.Unlock()
 151                         delete(sch.locking, uuid)
 152                 }()
 153                 err := sch.queue.Lock(uuid)
 154                 if err != nil {
 155                         logger.WithError(err).Warn("error locking container")
 156                         return
 157                 }
 158                 logger.Debug("lock succeeded")
 159                 ctr, ok := sch.queue.Get(uuid)
 160                 if !ok {
 161                         logger.Error("(BUG?) container disappeared from queue after Lock succeeded")
 162                 } else if ctr.State != arvados.ContainerStateLocked {
 163                         logger.Warnf("(race?) container has state=%q after Lock succeeded", ctr.State)
 164                 }
 165         }()
 166 }