-
Notifications
You must be signed in to change notification settings - Fork 8
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Cleanup running containers on the Control-C signal #422
base: master
Are you sure you want to change the base?
Conversation
Tested while multiple containers were running on all build hosts. |
54b622f
to
770ffb1
Compare
284f396
to
752d2fc
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I haven't done any testing yet but already found some potential issues in the code.
let running_jobs = jobs | ||
.into_iter() | ||
.map(|prep| { | ||
trace!(parent: &run_span, job_uuid = %prep.1.jobdef.job.uuid(), "Creating JobTask"); | ||
// the sender is set or we need to use the root sender | ||
let sender = prep | ||
.3 | ||
.into_inner() | ||
.unwrap_or_else(|| vec![root_sender.clone()]); | ||
JobTask::new(prep.0, prep.1, sender) | ||
}) | ||
.inspect( | ||
|task| trace!(parent: &run_span, job_uuid = %task.jobdef.job.uuid(), "Running job"), | ||
) | ||
.map(|task| { | ||
task.run() | ||
.instrument(tracing::debug_span!(parent: &run_span, "JobTask::run")) | ||
}) | ||
.collect::<futures::stream::FuturesUnordered<_>>(); | ||
debug!("Built {} jobs", running_jobs.len()); | ||
|
||
running_jobs | ||
.collect::<Result<()>>() | ||
.instrument(run_span.clone()) | ||
.await?; | ||
trace!(parent: &run_span, "All jobs finished"); | ||
drop(run_span); | ||
|
||
match root_receiver.recv().await { | ||
None => Err(anyhow!("No result received...")), | ||
Some(Ok(results)) => { | ||
let results = results | ||
.into_iter() | ||
.flat_map(|tpl| tpl.1.into_iter()) | ||
.map(ProducedArtifact::unpack) | ||
.collect(); | ||
Ok((results, HashMap::with_capacity(0))) | ||
|
||
tokio::select! { | ||
_ = token.cancelled() => { | ||
anyhow::bail!("Received Control-C signal"); | ||
} | ||
r = async { | ||
let running_jobs = jobs | ||
.into_iter() | ||
.map(|prep| { | ||
trace!(parent: &run_span, job_uuid = %prep.1.jobdef.job.uuid(), "Creating JobTask"); | ||
// the sender is set or we need to use the root sender | ||
let sender = prep | ||
.3 | ||
.into_inner() | ||
.unwrap_or_else(|| vec![root_sender.clone()]); | ||
JobTask::new(prep.0, prep.1, sender) | ||
}) | ||
.inspect( | ||
|task| trace!(parent: &run_span, job_uuid = %task.jobdef.job.uuid(), "Running job"), | ||
) | ||
.map(|task| { | ||
task.run() | ||
.instrument(tracing::debug_span!(parent: &run_span, "JobTask::run")) | ||
}) | ||
.collect::<futures::stream::FuturesUnordered<_>>(); | ||
debug!("Built {} jobs", running_jobs.len()); | ||
|
||
running_jobs | ||
.collect::<Result<()>>() | ||
.instrument(run_span.clone()) | ||
.await?; | ||
trace!(parent: &run_span, "All jobs finished"); | ||
drop(run_span); | ||
|
||
match root_receiver.recv().await { | ||
None => Err(anyhow!("No result received...")), | ||
Some(Ok(results)) => { | ||
let results = results | ||
.into_iter() | ||
.flat_map(|tpl| tpl.1.into_iter()) | ||
.map(ProducedArtifact::unpack) | ||
.collect(); | ||
Ok((results, HashMap::with_capacity(0))) | ||
} | ||
Some(Err(errors)) => Ok((vec![], errors)), | ||
} | ||
} => { | ||
r | ||
} | ||
Some(Err(errors)) => Ok((vec![], errors)), |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Note to self: I still need to properly review this part.
pub struct JobHandle { | ||
log_dir: Option<PathBuf>, | ||
endpoint: EndpointHandle, | ||
container_id: Option<String>, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This approach seems fine but we never set container_id
back to None
for finished jobs - it would be more elegant if we could do so (after ensuring that the container has indeed exited but that might already be implemented to check if the job has finished).
752d2fc
to
4642a14
Compare
- Add the `signal` feature to `tokio` to interrupt and handle the Control-C signal in Butido. - Add Control-C signal handling into the `Orchestrator`. - Implement `Drop` on the `JobHandle` to ensure container cleanup. Fixes science-computing#409 Signed-off-by: Nico Steinle <nico.steinle@eviden.com>
4642a14
to
9e52a33
Compare
signal
feature totokio
to interrupt and handle the Control-C signal in Butido.Orchestrator
.Drop
on theJobHandle
to ensure container cleanup.This is a working draft pr for testing purposes and still missing some features.