1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
|
"""
Note: slurm configuration for debian::
sudo apt-get install slurm-llnl
sudo vi /etc/slurm-llnl/slurm.conf
sudo mkdir /var/run/slurm-llnl
sudo chmod slurm:slurm /var/run/slurm-llnl
Now put it in rc.d::
sudo update-rc.d munge defaults
sudo update-rc.d slurm-llnl defaults
Or run the following each time::
sudo service munge start
sudo service slurm-llnl start
"""
import os
import subprocess
import multiprocessing
from .jobid import get_jobid
from . import store, runjob
# from park import config
# from park import environment
# Queue status words
_ACTIVE = ["RUNNING", "COMPLETING"]
_INACTIVE = ["PENDING", "SUSPENDED"]
_ERROR = ["CANCELLED", "FAILED", "TIMEOUT", "NODE_FAIL"]
_COMPLETE = ["COMPLETED"]
class Scheduler(object):
def jobs(self, status=None):
"""
Return a list of jobs on the queue.
"""
# TODO: list completed but not deactivated as well as completed
# print "queue"
output, _ = _slurm_exec("squeue", "-o", "%i %M %j")
# print "output",output
return output
def deactivate(self, jobid):
# TODO: remove the job from the active list
pass
# Job specific commands
def submit(self, request, origin):
"""
Put a command on batch queue, returning its job id.
"""
# print "submitting job",jobid
jobid = get_jobid()
store.create(jobid)
store.put(jobid, "request", request)
service = runjob.build_command(jobid, request)
num_workers = multiprocessing.cpu_count()
jobdir = store.path(jobid)
script = os.path.join(jobdir, "J%s.sh" % jobid)
# commands = ['export %s="%s"'%(k,v) for k,v in config.env().items()]
commands = ["srun -n 1 -K -o slurm-out.txt nice -n 19 %s &" % service]
# "srun -n %d -K -o kernel.out nice -n 19 %s"%(num_workers,kernel)]
create_batchfile(script, commands)
_out, err = _slurm_exec(
"sbatch",
"-n",
str(num_workers), # Number of tasks
#'-K', # Kill if any process returns error
#'-o', 'job%j.out', # output file
"-D",
jobdir, # Start directory
script,
)
if not err.startswith("sbatch: Submitted batch job "):
raise RuntimeError(err)
slurmid = err[28:].strip()
store.put(jobid, "slurmid", slurmid)
return jobid
def results(self, id):
try:
return runjob.results(id)
except KeyError:
pass
return {"status": self.status(id)}
def info(self, id):
request = store.get(id, "request")
request["id"] = id
return request
def status(self, id):
"""
Returns the follow states:
PENDING -- Job is waiting to be processed
ACTIVE -- Job is busy being processed through the queue
COMPLETE -- Job has completed successfully
ERROR -- Job has either been canceled by the user or an
error has been raised
"""
# Simple checks first
jobdir = store.path(id)
if not os.path.exists(jobdir):
return "UNKNOWN"
elif store.contains(id, "results"):
return "COMPLETE"
# Translate job id to slurm id
slurmid = store.get(id, "slurmid")
# Check slurm queue for job id
out, _ = _slurm_exec("squeue", "-h", "--format=%i %T")
out = out.strip()
state = ""
inqueue = False
if out != "":
for line in out.split("\n"):
line = line.split()
if slurmid == line[0]:
state = line[1]
inqueue = True
break
if inqueue:
if state in _ACTIVE:
return "ACTIVE"
elif state in _INACTIVE:
return "PENDING"
elif state in _COMPLETE:
return "COMPLETE"
elif state in _ERROR:
return "ERROR"
else:
raise RuntimeError("unexpected state from squeue: %s" % state)
else:
return "ERROR"
def cancel(self, jobid):
# print "canceling",jobid
slurmid = store.get(jobid, "slurmid")
_slurm_exec("scancel", slurmid)
def delete(self, jobid):
# jobdir = store.path(jobid)
self.cancel(jobid)
store.destroy(jobid)
def nextjob(self, queue):
raise NotImplementedError("SLURM queues do not support work sharing")
def postjob(self, id, results):
raise NotImplementedError("SLURM queues do not support work sharing")
def create_batchfile(script, commands):
"""
Create the batchfile to run the job.
"""
fid = open(script, "w")
fid.write("#!/bin/sh\n")
fid.write("\n".join(commands))
fid.write("\nwait\n")
fid.close()
return script
def _slurm_exec(cmd, *args):
"""
Run a slurm command, capturing any errors.
"""
# print "cmd",cmd,"args",args
process = subprocess.Popen([cmd] + list(args), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = process.communicate()
if err.startswith(cmd + ": error: "):
raise RuntimeError(cmd + ": " + err[15:].strip())
return out, err
|
