16314: Merge branch 'master'

[arvados.git] / lib / dispatchcloud / scheduler / run_queue.go

diff --git a/lib/dispatchcloud/scheduler/run_queue.go b/lib/dispatchcloud/scheduler/run_queue.go
index ecdae7f876786b948019b871d3018e3e42b48901..dddb974b326fbe7d61c280148e71f4f5c86e7abe 100644 (file)
--- a/lib/dispatchcloud/scheduler/run_queue.go
+++ b/lib/dispatchcloud/scheduler/run_queue.go
@@ -6,9 +6,10 @@ package scheduler
 
 import (
        "sort"
+       "time"
 
-       "git.curoverse.com/arvados.git/lib/dispatchcloud/container"
-       "git.curoverse.com/arvados.git/sdk/go/arvados"
+       "git.arvados.org/arvados.git/lib/dispatchcloud/container"
+       "git.arvados.org/arvados.git/sdk/go/arvados"
        "github.com/sirupsen/logrus"
 )
 
@@ -50,7 +51,7 @@ tryrun:
                                overquota = sorted[i:]
                                break tryrun
                        }
-                       sch.bgLock(logger, ctr.UUID)
+                       go sch.lockContainer(logger, ctr.UUID)
                        unalloc[it]--
                case arvados.ContainerStateLocked:
                        if unalloc[it] > 0 {
@@ -87,6 +88,8 @@ tryrun:
                                // a higher-priority container on the
                                // same instance type. Don't let this
                                // one sneak in ahead of it.
+                       } else if sch.pool.KillContainer(ctr.UUID, "about to lock") {
+                               logger.Info("not restarting yet: crunch-run process from previous attempt has not exited")
                        } else if sch.pool.StartContainer(it, ctr) {
                                // Success.
                        } else {
@@ -120,22 +123,16 @@ tryrun:
        }
 }
 
-// Start an API call to lock the given container, and return
-// immediately while waiting for the response in a new goroutine. Do
-// nothing if a lock request is already in progress for this
-// container.
-func (sch *Scheduler) bgLock(logger logrus.FieldLogger, uuid string) {
-       logger.Debug("locking")
-       sch.mtx.Lock()
-       defer sch.mtx.Unlock()
-       if sch.locking[uuid] {
-               logger.Debug("locking in progress, doing nothing")
+// Lock the given container. Should be called in a new goroutine.
+func (sch *Scheduler) lockContainer(logger logrus.FieldLogger, uuid string) {
+       if !sch.uuidLock(uuid, "lock") {
                return
        }
+       defer sch.uuidUnlock(uuid)
        if ctr, ok := sch.queue.Get(uuid); !ok || ctr.State != arvados.ContainerStateQueued {
                // This happens if the container has been cancelled or
                // locked since runQueue called sch.queue.Entries(),
-               // possibly by a bgLock() call from a previous
+               // possibly by a lockContainer() call from a previous
                // runQueue iteration. In any case, we will respond
                // appropriately on the next runQueue iteration, which
                // will have already been triggered by the queue
@@ -143,24 +140,50 @@ func (sch *Scheduler) bgLock(logger logrus.FieldLogger, uuid string) {
                logger.WithField("State", ctr.State).Debug("container no longer queued by the time we decided to lock it, doing nothing")
                return
        }
-       sch.locking[uuid] = true
-       go func() {
-               defer func() {
-                       sch.mtx.Lock()
-                       defer sch.mtx.Unlock()
-                       delete(sch.locking, uuid)
-               }()
-               err := sch.queue.Lock(uuid)
-               if err != nil {
-                       logger.WithError(err).Warn("error locking container")
-                       return
-               }
-               logger.Debug("lock succeeded")
-               ctr, ok := sch.queue.Get(uuid)
-               if !ok {
-                       logger.Error("(BUG?) container disappeared from queue after Lock succeeded")
-               } else if ctr.State != arvados.ContainerStateLocked {
-                       logger.Warnf("(race?) container has state=%q after Lock succeeded", ctr.State)
-               }
-       }()
+       err := sch.queue.Lock(uuid)
+       if err != nil {
+               logger.WithError(err).Warn("error locking container")
+               return
+       }
+       logger.Debug("lock succeeded")
+       ctr, ok := sch.queue.Get(uuid)
+       if !ok {
+               logger.Error("(BUG?) container disappeared from queue after Lock succeeded")
+       } else if ctr.State != arvados.ContainerStateLocked {
+               logger.Warnf("(race?) container has state=%q after Lock succeeded", ctr.State)
+       }
+}
+
+// Acquire a non-blocking lock for specified UUID, returning true if
+// successful.  The op argument is used only for debug logs.
+//
+// If the lock is not available, uuidLock arranges to wake up the
+// scheduler after a short delay, so it can retry whatever operation
+// is trying to get the lock (if that operation is still worth doing).
+//
+// This mechanism helps avoid spamming the controller/database with
+// concurrent updates for any single container, even when the
+// scheduler loop is running frequently.
+func (sch *Scheduler) uuidLock(uuid, op string) bool {
+       sch.mtx.Lock()
+       defer sch.mtx.Unlock()
+       logger := sch.logger.WithFields(logrus.Fields{
+               "ContainerUUID": uuid,
+               "Op":            op,
+       })
+       if op, locked := sch.uuidOp[uuid]; locked {
+               logger.Debugf("uuidLock not available, Op=%s in progress", op)
+               // Make sure the scheduler loop wakes up to retry.
+               sch.wakeup.Reset(time.Second / 4)
+               return false
+       }
+       logger.Debug("uuidLock acquired")
+       sch.uuidOp[uuid] = op
+       return true
+}
+
+func (sch *Scheduler) uuidUnlock(uuid string) {
+       sch.mtx.Lock()
+       defer sch.mtx.Unlock()
+       delete(sch.uuidOp, uuid)
 }