X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/bad6499910388c17c4a54b3128f361aa36a670e1..98c6c6990061c546b9995ad70766589499fb4844:/services/crunch-dispatch-slurm/crunch-dispatch-slurm.go

diff --git a/services/crunch-dispatch-slurm/crunch-dispatch-slurm.go b/services/crunch-dispatch-slurm/crunch-dispatch-slurm.go
index 44123f85cf..23e4b3a8cb 100644
--- a/services/crunch-dispatch-slurm/crunch-dispatch-slurm.go
+++ b/services/crunch-dispatch-slurm/crunch-dispatch-slurm.go
@@ -17,14 +17,16 @@ import (
 	"strings"
 	"time"
 
+	"git.curoverse.com/arvados.git/lib/dispatchcloud"
 	"git.curoverse.com/arvados.git/sdk/go/arvados"
 	"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
 	"git.curoverse.com/arvados.git/sdk/go/config"
 	"git.curoverse.com/arvados.git/sdk/go/dispatch"
-	"git.curoverse.com/arvados.git/services/dispatchcloud"
 	"github.com/coreos/go-systemd/daemon"
 )
 
+const initialNiceValue int64 = 10000
+
 var (
 	version           = "dev"
 	defaultConfigPath = "/etc/arvados/crunch-dispatch-slurm/crunch-dispatch-slurm.yml"
@@ -40,6 +42,7 @@ type Dispatcher struct {
 
 	SbatchArguments []string
 	PollPeriod      arvados.Duration
+	PrioritySpread  int64
 
 	// crunch-run command to invoke. The container UUID will be
 	// appended. If nil, []string{"crunch-run"} will be used.
@@ -47,6 +50,10 @@ type Dispatcher struct {
 	// Example: []string{"crunch-run", "--cgroup-parent-subsystem=memory"}
 	CrunchRunCommand []string
 
+	// Extra RAM to reserve (in Bytes) for SLURM job, in addition
+	// to the amount specified in the container's RuntimeConstraints
+	ReserveExtraRAM int64
+
 	// Minimum time between two attempts to run the same container
 	MinRetryPeriod arvados.Duration
 }
@@ -150,8 +157,9 @@ func (disp *Dispatcher) setup() {
 
 	disp.slurm = &slurmCLI{}
 	disp.sqCheck = &SqueueChecker{
-		Period: time.Duration(disp.PollPeriod),
-		Slurm:  disp.slurm,
+		Period:         time.Duration(disp.PollPeriod),
+		PrioritySpread: disp.PrioritySpread,
+		Slurm:          disp.slurm,
 	}
 	disp.Dispatcher = &dispatch.Dispatcher{
 		Arv:            arv,
@@ -194,19 +202,8 @@ func (disp *Dispatcher) checkSqueueForOrphans() {
 	}
 }
 
-func (disp *Dispatcher) niceness(priority int) int {
-	if priority > 1000 {
-		priority = 1000
-	}
-	if priority < 0 {
-		priority = 0
-	}
-	// Niceness range 1-10000
-	return (1000 - priority) * 10
-}
-
 func (disp *Dispatcher) sbatchArgs(container arvados.Container) ([]string, error) {
-	mem := int64(math.Ceil(float64(container.RuntimeConstraints.RAM+container.RuntimeConstraints.KeepCacheRAM) / float64(1048576)))
+	mem := int64(math.Ceil(float64(container.RuntimeConstraints.RAM+container.RuntimeConstraints.KeepCacheRAM+disp.ReserveExtraRAM) / float64(1048576)))
 
 	var disk int64
 	for _, m := range container.Mounts {
@@ -222,7 +219,7 @@ func (disp *Dispatcher) sbatchArgs(container arvados.Container) ([]string, error
 	sbatchArgs = append(sbatchArgs, fmt.Sprintf("--mem=%d", mem))
 	sbatchArgs = append(sbatchArgs, fmt.Sprintf("--cpus-per-task=%d", container.RuntimeConstraints.VCPUs))
 	sbatchArgs = append(sbatchArgs, fmt.Sprintf("--tmp=%d", disk))
-	sbatchArgs = append(sbatchArgs, fmt.Sprintf("--nice=%d", disp.niceness(container.Priority)))
+	sbatchArgs = append(sbatchArgs, fmt.Sprintf("--nice=%d", initialNiceValue))
 	if len(container.SchedulingParameters.Partitions) > 0 {
 		sbatchArgs = append(sbatchArgs, fmt.Sprintf("--partition=%s", strings.Join(container.SchedulingParameters.Partitions, ",")))
 	}
@@ -234,7 +231,7 @@ func (disp *Dispatcher) sbatchArgs(container arvados.Container) ([]string, error
 	} else if err != nil {
 		return nil, err
 	} else {
-		sbatchArgs = append(sbatchArgs, "--constraint="+it.Name)
+		sbatchArgs = append(sbatchArgs, "--constraint=instancetype="+it.Name)
 	}
 
 	return sbatchArgs, nil
@@ -323,12 +320,19 @@ func (disp *Dispatcher) runContainer(_ *dispatch.Dispatcher, ctr arvados.Contain
 				log.Printf("container %s has state %q, priority %d: cancel slurm job", ctr.UUID, updated.State, updated.Priority)
 				disp.scancel(ctr)
 			} else {
-				disp.renice(updated)
+				p := int64(updated.Priority)
+				if p <= 1000 {
+					// API is providing
+					// user-assigned priority. If
+					// ctrs have equal priority,
+					// run the older one first.
+					p = int64(p)<<50 - (updated.CreatedAt.UnixNano() >> 14)
+				}
+				disp.sqCheck.SetPriority(ctr.UUID, p)
 			}
 		}
 	}
 }
-
 func (disp *Dispatcher) scancel(ctr arvados.Container) {
 	disp.sqCheck.L.Lock()
 	err := disp.slurm.Cancel(ctr.UUID)
@@ -343,28 +347,6 @@ func (disp *Dispatcher) scancel(ctr arvados.Container) {
 	}
 }
 
-func (disp *Dispatcher) renice(ctr arvados.Container) {
-	nice := disp.niceness(ctr.Priority)
-	oldnice := disp.sqCheck.GetNiceness(ctr.UUID)
-	if nice == oldnice || oldnice == -1 {
-		return
-	}
-	log.Printf("updating slurm nice value to %d (was %d)", nice, oldnice)
-	disp.sqCheck.L.Lock()
-	err := disp.slurm.Renice(ctr.UUID, nice)
-	disp.sqCheck.L.Unlock()
-
-	if err != nil {
-		log.Printf("renice: %s", err)
-		time.Sleep(time.Second)
-		return
-	}
-	if disp.sqCheck.HasUUID(ctr.UUID) {
-		log.Printf("container %s has arvados priority %d, slurm nice %d",
-			ctr.UUID, ctr.Priority, disp.sqCheck.GetNiceness(ctr.UUID))
-	}
-}
-
 func (disp *Dispatcher) readConfig(path string) error {
 	err := config.LoadFile(disp, path)
 	if err != nil && os.IsNotExist(err) && path == defaultConfigPath {