X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/8af8ef37ba4f673e47ba17899a83e2a683872b55..e2cc6c1935565ee965f38501a4974e9d0e6b29e5:/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 f77023697e..d1f19dd7b5 100644 --- a/services/crunch-dispatch-slurm/crunch-dispatch-slurm.go +++ b/services/crunch-dispatch-slurm/crunch-dispatch-slurm.go @@ -7,6 +7,7 @@ package main // Dispatcher service for Crunch that submits containers to the slurm queue. import ( + "bytes" "context" "flag" "fmt" @@ -25,6 +26,8 @@ import ( "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 +43,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. @@ -154,8 +158,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, @@ -198,18 +203,7 @@ 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) { +func (disp *Dispatcher) slurmConstraintArgs(container arvados.Container) []string { mem := int64(math.Ceil(float64(container.RuntimeConstraints.RAM+container.RuntimeConstraints.KeepCacheRAM+disp.ReserveExtraRAM) / float64(1048576))) var disk int64 @@ -219,29 +213,36 @@ func (disp *Dispatcher) sbatchArgs(container arvados.Container) ([]string, error } } disk = int64(math.Ceil(float64(disk) / float64(1048576))) - - var sbatchArgs []string - sbatchArgs = append(sbatchArgs, disp.SbatchArguments...) - sbatchArgs = append(sbatchArgs, fmt.Sprintf("--job-name=%s", container.UUID)) - 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))) - if len(container.SchedulingParameters.Partitions) > 0 { - sbatchArgs = append(sbatchArgs, fmt.Sprintf("--partition=%s", strings.Join(container.SchedulingParameters.Partitions, ","))) + return []string{ + fmt.Sprintf("--mem=%d", mem), + fmt.Sprintf("--cpus-per-task=%d", container.RuntimeConstraints.VCPUs), + fmt.Sprintf("--tmp=%d", disk), } +} + +func (disp *Dispatcher) sbatchArgs(container arvados.Container) ([]string, error) { + var args []string + args = append(args, disp.SbatchArguments...) + args = append(args, "--job-name="+container.UUID, fmt.Sprintf("--nice=%d", initialNiceValue)) if disp.cluster == nil { // no instance types configured + args = append(args, disp.slurmConstraintArgs(container)...) } else if it, err := dispatchcloud.ChooseInstanceType(disp.cluster, &container); err == dispatchcloud.ErrInstanceTypesNotConfigured { // ditto + args = append(args, disp.slurmConstraintArgs(container)...) } else if err != nil { return nil, err } else { - sbatchArgs = append(sbatchArgs, "--constraint=instancetype="+it.Name) + // use instancetype constraint instead of slurm mem/cpu/tmp specs + args = append(args, "--constraint=instancetype="+it.Name) } - return sbatchArgs, nil + if len(container.SchedulingParameters.Partitions) > 0 { + args = append(args, "--partition="+strings.Join(container.SchedulingParameters.Partitions, ",")) + } + + return args, nil } func (disp *Dispatcher) submit(container arvados.Container, crunchRunCommand []string) error { @@ -274,8 +275,21 @@ func (disp *Dispatcher) runContainer(_ *dispatch.Dispatcher, ctr arvados.Contain log.Printf("Submitting container %s to slurm", ctr.UUID) if err := disp.submit(ctr, disp.CrunchRunCommand); err != nil { var text string - if err == dispatchcloud.ErrConstraintsNotSatisfiable { - text = fmt.Sprintf("cannot run container %s: %s", ctr.UUID, err) + if err, ok := err.(dispatchcloud.ConstraintsNotSatisfiableError); ok { + var logBuf bytes.Buffer + fmt.Fprintf(&logBuf, "cannot run container %s: %s\n", ctr.UUID, err) + if len(err.AvailableTypes) == 0 { + fmt.Fprint(&logBuf, "No instance types are configured.\n") + } else { + fmt.Fprint(&logBuf, "Available instance types:\n") + for _, t := range err.AvailableTypes { + fmt.Fprintf(&logBuf, + "Type %q: %d VCPUs, %d RAM, %d Scratch, %f Price\n", + t.Name, t.VCPUs, t.RAM, t.Scratch, t.Price, + ) + } + } + text = logBuf.String() disp.UpdateState(ctr.UUID, dispatch.Cancelled) } else { text = fmt.Sprintf("Error submitting container %s to slurm: %s", ctr.UUID, err) @@ -327,12 +341,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) @@ -347,28 +368,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 {