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
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
use std::{
collections::HashMap,
sync::{
atomic::{AtomicBool, AtomicU64, Ordering},
Arc, Mutex,
},
};
use regex::Regex;
use crate::ExpandedDescr;
use super::*;
pub struct SlurmExecutorBuilder {
pub max_jobs: u64,
}
impl ExecutorBuilder for SlurmExecutorBuilder {
type Executor = SlurmExecutor;
fn init<J: JobUnit>(
self,
) -> (
Self::Executor,
Sender<ToRun<J>>,
Sender<ExecutorResult>,
Receiver<ExecutorResult>,
) {
let to_run_chan = crossbeam_channel::unbounded::<ToRun<J>>();
let done_chan = crossbeam_channel::unbounded();
let submitted_jobs = Arc::new(Mutex::new(HashMap::new()));
let done_submitting = Arc::new(AtomicBool::new(false));
let currently_submitted = Arc::new(AtomicU64::new(0));
// TODO: max submission at at time
let mut submitting_threads = Vec::new();
for _i in 0..32 {
let submitting_thread = {
let rcvr = to_run_chan.1.clone();
let done_sndr = done_chan.0.clone();
let submitted_jobs_ptr = submitted_jobs.clone();
let done_submitting = done_submitting.clone();
let currently_submitted = currently_submitted.clone();
let max_jobs = self.max_jobs;
thread::Builder::new()
.name("submitting_thread".into())
.spawn(move || {
for to_run in rcvr.into_iter() {
while currently_submitted.load(Ordering::Acquire) >= max_jobs {
thread::park();
}
let j_idx = to_run.job_idx;
let stdout = io::stdout();
{
let mut lock = stdout.lock();
writeln!(lock, "Submitting job: {:?}", to_run.job_unit).unwrap();
write!(lock, "Submitting job: {:#?} interpreted as: {:#?} ", to_run.job_unit, ExpandedDescr(to_run.job_unit())).unwrap();
writeln!(lock, "\n").unwrap();
}
super::create_out_dir(&to_run.job_unit);
let mut sbatch_cmd = Command::new("sbatch");
sbatch_cmd.arg("-J")
.arg(to_run.job_unit.name())
.arg("--partition=common,dedicated,inbio")
// .arg("--partition=inbio")
.arg("--exclude=maestro-1101")
.arg("--qos=fast")
.arg(format!("--nice={}",to_run.job_unit.nice()))
.arg("--parsable");
let sbatch_cmd = match to_run.job_unit.log_file() {
Some(p) => {sbatch_cmd.arg("--output").arg(p)}
None => {sbatch_cmd.arg("--output=slurm-log-snakemake.out")
.arg("--open-mode=append")}
}.arg("--wrap")
.arg(to_run.job_unit.cmd().unwrap())
.output()
.unwrap();
if sbatch_cmd.status.success() {
currently_submitted.fetch_add(1, Ordering::AcqRel);
let slurm_job_id: u64 = std::str::from_utf8(&sbatch_cmd.stdout)
.unwrap()
.trim()
.parse()
.unwrap();
submitted_jobs_ptr
.lock()
.unwrap()
.insert(slurm_job_id, j_idx);
} else {
done_sndr
.send(ExecutorResult {
job_idx: j_idx,
result: Err(format!(
"Could not queue job. Exited with status {:?} and stderr:\n{}",
sbatch_cmd.status.code(),
String::from_utf8_lossy(&sbatch_cmd.stderr)
)),
})
.unwrap();
}
}
done_submitting.store(true, Ordering::Release);
})
.unwrap()
};
submitting_threads.push(submitting_thread);
}
let status_polling_thread = {
let sndr = done_chan.0.clone();
let submitting_threads: Vec<_> = submitting_threads
.iter()
.map(|jh| jh.thread().clone())
.collect();
let currently_submitted = currently_submitted;
thread::Builder::new()
.name("status_polling_thread".into())
.spawn(move || {
let stdout = io::stdout();
let re = Regex::new(r"(\d+)(\.[^\|]+)?\|([^\|]+)\|([^\|]+)\|").unwrap();
let mut rate_limiter = fence::Fence::from_millis(100);
loop {
let jobs_slurm_ids: Vec<_> = submitted_jobs
.lock()
.unwrap()
.keys()
.map(|j| j.to_string())
.collect();
if jobs_slurm_ids.is_empty() {
if done_submitting.load(Ordering::Acquire) {
break;
}
} else {
let ids_to_query = jobs_slurm_ids.join(",");
dbg!(&ids_to_query);
let sacct_cmd = Command::new("sacct")
.arg("-j")
.arg(ids_to_query)
.arg("--noheader")
.arg("--parsable")
.arg("--format=JobId,State,Elapsed")
.output()
.unwrap();
if !sacct_cmd.status.success() {
panic!()
}
// Consider that all jobs will indeed be present
for l in std::str::from_utf8(&sacct_cmd.stdout).unwrap().lines() {
let cap = re.captures(l).unwrap();
if cap.get(2).is_some() {
continue;
}
let jobid = cap.get(1).unwrap().as_str();
let job_status = cap.get(3).unwrap().as_str();
let elapsed = cap.get(4).unwrap().as_str();
match job_status {
"PENDING" | "CONFIGURING" | "COMPLETING" | "RUNNING"
| "SUSPENDED" | "PREEMPTED" => {
//do nothing, the job is still running
}
"COMPLETED" => {
let graph_idx = submitted_jobs
.lock()
.unwrap()
.remove(&jobid.parse().unwrap())
.unwrap();
{
let mut lock = stdout.lock();
writeln!(
lock,
"Finished job: {:?}. Elapsed time: {}\n",
jobid, elapsed
)
.unwrap();
}
sndr.send(ExecutorResult {
job_idx: graph_idx,
result: Ok(()),
})
.unwrap();
submitted_jobs
.lock()
.unwrap()
.remove(&jobid.parse().unwrap());
currently_submitted.fetch_sub(1, Ordering::Acquire);
for t in &submitting_threads {
t.unpark();
}
}
s => {
let graph_idx = submitted_jobs
.lock()
.unwrap()
.remove(&jobid.parse().unwrap())
.unwrap();
sndr.send(ExecutorResult {
job_idx: graph_idx,
result: Err(format!(
"Execution failed on the cluster. Slurm jobid: {}. State: {}",
jobid,
s
)),
})
.unwrap();
submitted_jobs
.lock()
.unwrap()
.remove(&jobid.parse().unwrap());
}
}
}
}
rate_limiter.sleep();
}
})
.unwrap()
};
(
SlurmExecutor {
submitting_threads,
status_polling_thread,
},
to_run_chan.0,
done_chan.0,
done_chan.1,
)
}
}
pub struct SlurmExecutor {
submitting_threads: Vec<JoinHandle<()>>,
status_polling_thread: JoinHandle<()>,
}
impl Executor for SlurmExecutor {
fn join(self) -> Vec<std::thread::Result<()>> {
let mut res = Vec::with_capacity(self.submitting_threads.len() + 1);
res.push(self.status_polling_thread.join());
res.extend(self.submitting_threads.into_iter().map(|jh| jh.join()));
res
}
}