Merge pull request #1652 from Kobzol/timeout-async

Add compilation benchmark timeout

Author: Kobzol

Cargo.lock

@@ -21,7 +21,7 @@ semver = "1.0"
 reqwest = { version = "0.11", features = ["json"] }
 xz2 = "0.1.3"
 tar = "0.4"
-tokio = { version = "1.6", features = ["rt"] }
+tokio = { version = "1.6", features = ["rt", "process"] }
 database = { path = "../database" }
 intern = { path = "../intern" }
 futures = "0.3.5"

collector/Cargo.toml

@@ -17,11 +17,13 @@ use std::cmp::Ordering;
 use std::ffi::OsStr;
 use std::fs;
 use std::fs::File;
+use std::future::Future;
 use std::io::BufWriter;
 use std::io::Write;
 use std::path::{Path, PathBuf};
 use std::process;
 use std::process::{Command, Stdio};
+use std::time::Duration;
 use std::{str, time::Instant};
 use tokio::runtime::Runtime;
 
@@ -34,6 +36,7 @@ use collector::runtime::{
 use collector::toolchain::{
     create_toolchain_from_published_version, get_local_toolchain, Sysroot, Toolchain,
 };
+use collector::utils::wait_for_future;
 
 fn n_normal_benchmarks_remaining(n: usize) -> String {
     let suffix = if n == 1 { "" } else { "s" };
@@ -262,7 +265,13 @@ fn profile(
             let benchmark_id = format!("{} ({}/{})", benchmark.name, i + 1, benchmarks.len());
             eprintln!("Executing benchmark {benchmark_id}");
             let mut processor = ProfileProcessor::new(profiler, out_dir, &toolchain.id);
-            let result = benchmark.measure(&mut processor, profiles, scenarios, toolchain, Some(1));
+            let result = wait_for_future(benchmark.measure(
+                &mut processor,
+                profiles,
+                scenarios,
+                toolchain,
+                Some(1),
+            ));
             eprintln!("Finished benchmark {benchmark_id}");
 
             if let Err(ref s) = result {
@@ -611,6 +620,7 @@ fn main_result() -> anyhow::Result<i32> {
     builder
         .worker_threads(1)
         .max_blocking_threads(1)
+        .enable_time()
         .enable_io();
     let mut rt = builder.build().expect("built runtime");
 
@@ -1077,6 +1087,18 @@ fn bench_published_artifact(
     )
 }
 
+const COMPILE_BENCHMARK_TIMEOUT: Duration = Duration::from_secs(60 * 30);
+
+async fn with_timeout<F: Future<Output = anyhow::Result<()>>>(fut: F) -> anyhow::Result<()> {
+    match tokio::time::timeout(COMPILE_BENCHMARK_TIMEOUT, fut).await {
+        Ok(res) => res,
+        Err(_) => Err(anyhow::anyhow!(
+            "Benchmark timeouted in {} seconds",
+            COMPILE_BENCHMARK_TIMEOUT.as_secs()
+        )),
+    }
+}
+
 /// Perform compile benchmarks.
 fn bench_compile(
     rt: &mut Runtime,
@@ -1121,7 +1143,6 @@ fn bench_compile(
             print_intro();
 
             let mut processor = BenchProcessor::new(
-                rt,
                 tx.conn(),
                 benchmark_name,
                 &shared.artifact_id,
@@ -1157,13 +1178,14 @@ fn bench_compile(
                 )
             },
             &|processor| {
-                benchmark.measure(
+                rt.block_on(with_timeout(benchmark.measure(
                     processor,
                     &config.profiles,
                     &config.scenarios,
                     &shared.toolchain,
                     config.iterations,
-                )
+                )))
+                .with_context(|| anyhow::anyhow!("Cannot compile {}", benchmark.name))
             },
         )
     }
@@ -1175,8 +1197,7 @@ fn bench_compile(
             Category::Primary,
             &|| eprintln!("Special benchmark commencing (due to `--bench-rustc`)"),
             &|processor| {
-                processor
-                    .measure_rustc(&shared.toolchain)
+                rt.block_on(with_timeout(processor.measure_rustc(&shared.toolchain)))
                     .context("measure rustc")
             },
         );

collector/src/bin/collector.rs

@@ -4,6 +4,7 @@ use crate::compile::benchmark::profile::Profile;
 use crate::compile::benchmark::scenario::Scenario;
 use crate::compile::execute::{CargoProcess, Processor};
 use crate::toolchain::Toolchain;
+use crate::utils::wait_for_future;
 use anyhow::{bail, Context};
 use log::debug;
 use std::collections::{HashMap, HashSet};
@@ -189,7 +190,7 @@ impl Benchmark {
     }
 
     /// Run a specific benchmark under a processor + profiler combination.
-    pub fn measure(
+    pub async fn measure(
         &self,
         processor: &mut dyn Processor,
         profiles: &[Profile],
@@ -261,9 +262,11 @@ impl Benchmark {
             for (profile, prep_dir) in &profile_dirs {
                 let server = server.clone();
                 let thread = s.spawn::<_, anyhow::Result<()>>(move || {
-                    self.mk_cargo_process(toolchain, prep_dir.path(), *profile)
-                        .jobserver(server)
-                        .run_rustc(false)?;
+                    wait_for_future(
+                        self.mk_cargo_process(toolchain, prep_dir.path(), *profile)
+                            .jobserver(server)
+                            .run_rustc(false),
+                    )?;
                     Ok(())
                 });
                 threads.push(thread);
@@ -310,7 +313,8 @@ impl Benchmark {
                 if scenarios.contains(&Scenario::Full) {
                     self.mk_cargo_process(toolchain, cwd, profile)
                         .processor(processor, Scenario::Full, "Full", None)
-                        .run_rustc(true)?;
+                        .run_rustc(true)
+                        .await?;
                 }
 
                 // Rustdoc does not support incremental compilation
@@ -321,15 +325,17 @@ impl Benchmark {
                         self.mk_cargo_process(toolchain, cwd, profile)
                             .incremental(true)
                             .processor(processor, Scenario::IncrFull, "IncrFull", None)
-                            .run_rustc(true)?;
+                            .run_rustc(true)
+                            .await?;
                     }
 
                     // An incremental build with no changes (fastest incremental case).
                     if scenarios.contains(&Scenario::IncrUnchanged) {
                         self.mk_cargo_process(toolchain, cwd, profile)
                             .incremental(true)
                             .processor(processor, Scenario::IncrUnchanged, "IncrUnchanged", None)
-                            .run_rustc(true)?;
+                            .run_rustc(true)
+                            .await?;
                     }
 
                     if scenarios.contains(&Scenario::IncrPatched) {
@@ -348,7 +354,8 @@ impl Benchmark {
                                     &scenario_str,
                                     Some(patch),
                                 )
-                                .run_rustc(true)?;
+                                .run_rustc(true)
+                                .await?;
                         }
                     }
                 }