X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/03f277bf4b616f41ef7ed4b195d44dab83d16144..fd4d26e448b1f9f45f84c0ff9ec10db54471d880:/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 0879de20f9..c1009a5d8e 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. @@ -47,8 +51,15 @@ 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 + + // Batch size for container queries + BatchSize int64 } func main() { @@ -150,11 +161,13 @@ 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, + BatchSize: disp.BatchSize, RunContainer: disp.runContainer, PollPeriod: time.Duration(disp.PollPeriod), MinRetryPeriod: time.Duration(disp.MinRetryPeriod), @@ -194,19 +207,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))) +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 for _, m := range container.Mounts { @@ -215,29 +217,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 { @@ -270,8 +279,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) @@ -323,12 +345,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 +372,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 {