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/cloud"
  11         "git.curoverse.com/arvados.git/lib/dispatchcloud/container"
  12         "git.curoverse.com/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                         sch.bgLock(logger, ctr.UUID)
  55                         unalloc[it]--
  56                 case arvados.ContainerStateLocked:
  57                         if unalloc[it] < 1 {
  58                                 if sch.pool.AtQuota() {
  59                                         logger.Debug("not starting: AtQuota and no unalloc workers")
  60                                         overquota = sorted[i:]
  61                                         break tryrun
  62                                 }
  63                                 logger.Info("creating new instance")
  64                                 err := sch.pool.Create(it)
  65                                 if err != nil {
  66                                         if _, ok := err.(cloud.QuotaError); !ok {
  67                                                 logger.WithError(err).Warn("error creating worker")
  68                                         }
  69                                         sch.queue.Unlock(ctr.UUID)
  70                                         // Don't let lower-priority
  71                                         // containers starve this one
  72                                         // by using keeping idle
  73                                         // workers alive on different
  74                                         // instance types.  TODO:
  75                                         // avoid getting starved here
  76                                         // if instances of a specific
  77                                         // type always fail.
  78                                         overquota = sorted[i:]
  79                                         break tryrun
  80                                 }
  81                                 unalloc[it]++
  82                         }
  83
  84                         if dontstart[it] {
  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) {
  90                                 unalloc[it]--
  91                         } else {
  92                                 dontstart[it] = true
  93                         }
  94                 }
  95         }
  96
  97         if len(overquota) > 0 {
  98                 // Unlock any containers that are unmappable while
  99                 // we're at quota.
 100                 for _, ctr := range overquota {
 101                         ctr := ctr.Container
 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)
 106                                 if err != nil {
 107                                         logger.WithError(err).Warn("error unlocking")
 108                                 }
 109                         }
 110                 }
 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 {
 114                         if n < 1 {
 115                                 continue
 116                         }
 117                         sch.pool.Shutdown(it)
 118                 }
 119         }
 120 }
 121
 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
 125 // container.
 126 func (sch *Scheduler) bgLock(logger logrus.FieldLogger, uuid string) {
 127         logger.Debug("locking")
 128         sch.mtx.Lock()
 129         defer sch.mtx.Unlock()
 130         if sch.locking[uuid] {
 131                 logger.Debug("locking in progress, doing nothing")
 132                 return
 133         }
 134         sch.locking[uuid] = true
 135         go func() {
 136                 defer func() {
 137                         sch.mtx.Lock()
 138                         defer sch.mtx.Unlock()
 139                         delete(sch.locking, uuid)
 140                 }()
 141                 err := sch.queue.Lock(uuid)
 142                 if err != nil {
 143                         logger.WithError(err).Warn("error locking container")
 144                         return
 145                 }
 146                 ctr, ok := sch.queue.Get(uuid)
 147                 if !ok {
 148                         logger.Error("(BUG?) container disappeared from queue after Lock succeeded")
 149                 } else if ctr.State != arvados.ContainerStateLocked {
 150                         logger.Warnf("(race?) container has state=%q after Lock succeeded", ctr.State)
 151                 }
 152         }()
 153 }