1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
17 const slurm15NiceLimit int64 = 10000
19 type slurmJob struct {
22 priority int64 // current slurm priority (incorporates nice value)
23 nice int64 // current slurm nice value
27 // Squeue implements asynchronous polling monitor of the SLURM queue using the
29 type SqueueChecker struct {
33 queue map[string]*slurmJob
40 // HasUUID checks if a given container UUID is in the slurm queue.
41 // This does not run squeue directly, but instead blocks until woken
42 // up by next successful update of squeue.
43 func (sqc *SqueueChecker) HasUUID(uuid string) bool {
44 sqc.startOnce.Do(sqc.start)
47 defer sqc.lock.RUnlock()
49 // block until next squeue broadcast signaling an update.
51 _, exists := sqc.queue[uuid]
55 // SetPriority sets or updates the desired (Arvados) priority for a
57 func (sqc *SqueueChecker) SetPriority(uuid string, want int64) {
58 sqc.startOnce.Do(sqc.start)
61 job := sqc.queue[uuid]
63 // Wait in case the slurm job was just submitted and
64 // will appear in the next squeue update.
68 needUpdate := job != nil && job.wantPriority != want
73 job.wantPriority = want
78 // adjust slurm job nice values as needed to ensure slurm priority
79 // order matches Arvados priority order.
80 func (sqc *SqueueChecker) reniceAll() {
82 defer sqc.lock.RUnlock()
84 jobs := make([]*slurmJob, 0, len(sqc.queue))
85 for _, j := range sqc.queue {
86 if j.wantPriority == 0 {
87 // SLURM job with unknown Arvados priority
88 // (perhaps it's not an Arvados job)
91 if j.priority <= 2*slurm15NiceLimit {
92 // SLURM <= 15.x implements "hold" by setting
93 // priority to 0. If we include held jobs
94 // here, we'll end up trying to push other
95 // jobs below them using negative priority,
96 // which won't help anything.
99 jobs = append(jobs, j)
102 sort.Slice(jobs, func(i, j int) bool {
103 if jobs[i].wantPriority != jobs[j].wantPriority {
104 return jobs[i].wantPriority > jobs[j].wantPriority
106 // break ties with container uuid --
107 // otherwise, the ordering would change from
108 // one interval to the next, and we'd do many
109 // pointless slurm queue rearrangements.
110 return jobs[i].uuid > jobs[j].uuid
113 renice := wantNice(jobs, sqc.PrioritySpread)
114 for i, job := range jobs {
116 if job.hitNiceLimit && niceNew > slurm15NiceLimit {
117 niceNew = slurm15NiceLimit
119 if niceNew == job.nice {
122 err := sqc.Slurm.Renice(job.uuid, niceNew)
123 if err != nil && niceNew > slurm15NiceLimit && strings.Contains(err.Error(), "Invalid nice value") {
124 log.Printf("container %q clamping nice values at %d, priority order will not be correct -- see https://dev.arvados.org/projects/arvados/wiki/SLURM_integration#Limited-nice-values-SLURM-15", job.uuid, slurm15NiceLimit)
125 job.hitNiceLimit = true
130 // Stop stops the squeue monitoring goroutine. Do not call HasUUID
131 // after calling Stop.
132 func (sqc *SqueueChecker) Stop() {
138 // check gets the names of jobs in the SLURM queue (running and
139 // queued). If it succeeds, it updates sqc.queue and wakes up any
140 // goroutines that are waiting in HasUUID() or All().
141 func (sqc *SqueueChecker) check() {
143 defer sqc.lock.Unlock()
145 cmd := sqc.Slurm.QueueCommand([]string{"--all", "--noheader", "--format=%j %y %Q %T %r"})
146 stdout, stderr := &bytes.Buffer{}, &bytes.Buffer{}
147 cmd.Stdout, cmd.Stderr = stdout, stderr
148 if err := cmd.Run(); err != nil {
149 log.Printf("Error running %q %q: %s %q", cmd.Path, cmd.Args, err, stderr.String())
153 lines := strings.Split(stdout.String(), "\n")
154 newq := make(map[string]*slurmJob, len(lines))
155 for _, line := range lines {
159 var uuid, state, reason string
161 if _, err := fmt.Sscan(line, &uuid, &n, &p, &state, &reason); err != nil {
162 log.Printf("warning: ignoring unparsed line in squeue output: %q", line)
165 replacing, ok := sqc.queue[uuid]
167 replacing = &slurmJob{uuid: uuid}
169 replacing.priority = p
171 newq[uuid] = replacing
173 if state == "PENDING" && ((reason == "BadConstraints" && p <= 2*slurm15NiceLimit) || reason == "launch failed requeued held") && replacing.wantPriority > 0 {
174 // When using SLURM 14.x or 15.x, our queued
175 // jobs land in this state when "scontrol
176 // reconfigure" invalidates their feature
177 // constraints by clearing all node features.
178 // They stay in this state even after the
179 // features reappear, until we run "scontrol
180 // release {jobid}". Priority is usually 0 in
181 // this state, but sometimes (due to a race
182 // with nice adjustments?) it's a small
185 // "scontrol release" is silent and successful
186 // regardless of whether the features have
187 // reappeared, so rather than second-guessing
188 // whether SLURM is ready, we just keep trying
189 // this until it works.
191 // "launch failed requeued held" seems to be
192 // another manifestation of this problem,
193 // resolved the same way.
194 log.Printf("releasing held job %q (priority=%d, state=%q, reason=%q)", uuid, p, state, reason)
195 sqc.Slurm.Release(uuid)
196 } else if state != "RUNNING" && p <= 2*slurm15NiceLimit && replacing.wantPriority > 0 {
197 log.Printf("warning: job %q has low priority %d, nice %d, state %q, reason %q", uuid, p, n, state, reason)
201 sqc.notify.Broadcast()
204 // Initialize, and start a goroutine to call check() once per
205 // squeue.Period until terminated by calling Stop().
206 func (sqc *SqueueChecker) start() {
207 sqc.notify.L = sqc.lock.RLocker()
208 sqc.done = make(chan struct{})
210 ticker := time.NewTicker(sqc.Period)
221 // If this iteration took
222 // longer than sqc.Period,
223 // consume the next tick and
224 // wait. Otherwise we would
225 // starve other goroutines.
233 // All waits for the next squeue invocation, and returns all job
234 // names reported by squeue.
235 func (sqc *SqueueChecker) All() []string {
236 sqc.startOnce.Do(sqc.start)
238 defer sqc.lock.RUnlock()
241 for u := range sqc.queue {
242 uuids = append(uuids, u)