X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/71fd4da18b22100682ae7e2079aadfd66360d310..215d15bc03a38be1965a5d9df21417a3f7eae032:/lib/dispatchcloud/container/queue.go

diff --git a/lib/dispatchcloud/container/queue.go b/lib/dispatchcloud/container/queue.go
index 847fe9e27c..4e807a12ab 100644
--- a/lib/dispatchcloud/container/queue.go
+++ b/lib/dispatchcloud/container/queue.go
@@ -5,6 +5,7 @@
 package container
 
 import (
+	"errors"
 	"io"
 	"sync"
 	"time"
@@ -131,7 +132,7 @@ func (cq *Queue) Forget(uuid string) {
 	defer cq.mtx.Unlock()
 	ctr := cq.current[uuid].Container
 	if ctr.State == arvados.ContainerStateComplete || ctr.State == arvados.ContainerStateCancelled {
-		delete(cq.current, uuid)
+		cq.delEnt(uuid, ctr.State)
 	}
 }
 
@@ -196,7 +197,7 @@ func (cq *Queue) Update() error {
 			cq.current[uuid] = cur
 		}
 	}
-	for uuid := range cq.current {
+	for uuid, ent := range cq.current {
 		if _, dontupdate := cq.dontupdate[uuid]; dontupdate {
 			// Don't expunge an entry that was
 			// added/updated locally after we started
@@ -207,7 +208,7 @@ func (cq *Queue) Update() error {
 			// the poll response (evidently it's
 			// cancelled, completed, deleted, or taken by
 			// a different dispatcher).
-			delete(cq.current, uuid)
+			cq.delEnt(uuid, ent.Container.State)
 		}
 	}
 	cq.dontupdate = nil
@@ -216,6 +217,15 @@ func (cq *Queue) Update() error {
 	return nil
 }
 
+// Caller must have lock.
+func (cq *Queue) delEnt(uuid string, state arvados.ContainerState) {
+	cq.logger.WithFields(logrus.Fields{
+		"ContainerUUID": uuid,
+		"State":         state,
+	}).Info("dropping container from queue")
+	delete(cq.current, uuid)
+}
+
 func (cq *Queue) addEnt(uuid string, ctr arvados.Container) {
 	it, err := cq.chooseType(&ctr)
 	if err != nil && (ctr.State == arvados.ContainerStateQueued || ctr.State == arvados.ContainerStateLocked) {
@@ -223,8 +233,25 @@ func (cq *Queue) addEnt(uuid string, ctr arvados.Container) {
 		// error: it wouldn't help to try again, or to leave
 		// it for a different dispatcher process to attempt.
 		errorString := err.Error()
-		cq.logger.WithField("ContainerUUID", ctr.UUID).Warn("cancel container with no suitable instance type")
+		logger := cq.logger.WithField("ContainerUUID", ctr.UUID)
+		logger.WithError(err).Warn("cancel container with no suitable instance type")
 		go func() {
+			if ctr.State == arvados.ContainerStateQueued {
+				// Can't set runtime error without
+				// locking first. If Lock() is
+				// successful, it will call addEnt()
+				// again itself, and we'll fall
+				// through to the
+				// setRuntimeError/Cancel code below.
+				err := cq.Lock(ctr.UUID)
+				if err != nil {
+					logger.WithError(err).Warn("lock failed")
+					// ...and try again on the
+					// next Update, if the problem
+					// still exists.
+				}
+				return
+			}
 			var err error
 			defer func() {
 				if err == nil {
@@ -239,14 +266,8 @@ func (cq *Queue) addEnt(uuid string, ctr arvados.Container) {
 				if latest.State == arvados.ContainerStateCancelled {
 					return
 				}
-				cq.logger.WithField("ContainerUUID", ctr.UUID).WithError(err).Warn("error while trying to cancel unsatisfiable container")
+				logger.WithError(err).Warn("error while trying to cancel unsatisfiable container")
 			}()
-			if ctr.State == arvados.ContainerStateQueued {
-				err = cq.Lock(ctr.UUID)
-				if err != nil {
-					return
-				}
-			}
 			err = cq.setRuntimeError(ctr.UUID, errorString)
 			if err != nil {
 				return
@@ -258,6 +279,12 @@ func (cq *Queue) addEnt(uuid string, ctr arvados.Container) {
 		}()
 		return
 	}
+	cq.logger.WithFields(logrus.Fields{
+		"ContainerUUID": ctr.UUID,
+		"State":         ctr.State,
+		"Priority":      ctr.Priority,
+		"InstanceType":  it.Name,
+	}).Info("adding container to queue")
 	cq.current[uuid] = QueueEnt{Container: ctr, InstanceType: it}
 }
 
@@ -372,32 +399,62 @@ func (cq *Queue) poll() (map[string]*arvados.Container, error) {
 	}
 	apply(avail)
 
-	var missing []string
+	missing := map[string]bool{}
 	cq.mtx.Lock()
 	for uuid, ent := range cq.current {
 		if next[uuid] == nil &&
 			ent.Container.State != arvados.ContainerStateCancelled &&
 			ent.Container.State != arvados.ContainerStateComplete {
-			missing = append(missing, uuid)
+			missing[uuid] = true
 		}
 	}
 	cq.mtx.Unlock()
 
-	for i, page := 0, 20; i < len(missing); i += page {
-		batch := missing[i:]
-		if len(batch) > page {
-			batch = batch[:page]
+	for len(missing) > 0 {
+		var batch []string
+		for uuid := range missing {
+			batch = append(batch, uuid)
+			if len(batch) == 20 {
+				break
+			}
 		}
+		filters := []arvados.Filter{{"uuid", "in", batch}}
 		ended, err := cq.fetchAll(arvados.ResourceListParams{
 			Select:  selectParam,
 			Order:   "uuid",
 			Count:   "none",
-			Filters: []arvados.Filter{{"uuid", "in", batch}},
+			Filters: filters,
 		})
 		if err != nil {
 			return nil, err
 		}
 		apply(ended)
+		if len(ended) == 0 {
+			// This is the only case where we can conclude
+			// a container has been deleted from the
+			// database. A short (but non-zero) page, on
+			// the other hand, can be caused by a response
+			// size limit.
+			for _, uuid := range batch {
+				cq.logger.WithField("ContainerUUID", uuid).Warn("container not found by controller (deleted?)")
+				delete(missing, uuid)
+				cq.mtx.Lock()
+				cq.delEnt(uuid, cq.current[uuid].Container.State)
+				cq.mtx.Unlock()
+			}
+			continue
+		}
+		for _, ctr := range ended {
+			if _, ok := missing[ctr.UUID]; !ok {
+				msg := "BUG? server response did not match requested filters, erroring out rather than risk deadlock"
+				cq.logger.WithFields(logrus.Fields{
+					"ContainerUUID": ctr.UUID,
+					"Filters":       filters,
+				}).Error(msg)
+				return nil, errors.New(msg)
+			}
+			delete(missing, ctr.UUID)
+		}
 	}
 	return next, nil
 }