Make collector call home to perf server

Processing now calls home so that we can display the current status on
the status page.

Author: Mark-Simulacrum

Cargo.lock

@@ -18,6 +18,7 @@ cargo_metadata = "0.5"
 lazy_static = "1"
 semver = "0.9"
 reqwest = "0.8.6"
+futures = "0.1"
 
 [dependencies.rustup]
 git = "https://github.com/rust-lang-nursery/rustup.rs.git"

collector/Cargo.toml

@@ -0,0 +1,22 @@
+pub mod collected {
+    use Commit;
+
+    #[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
+    pub enum Request {
+        // Will benchmark commit with these benchmarks
+        BenchmarkCommit {
+            commit: Commit,
+            benchmarks: Vec<String>,
+        },
+        // benchmark finished for this benchmark/commit
+        BenchmarkDone {
+            benchmark: String,
+            commit: Commit,
+        }
+    }
+
+    #[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
+    pub struct Response {
+        // nothing
+    }
+}

collector/src/api.rs

@@ -21,6 +21,9 @@ extern crate tempfile;
 extern crate rustup;
 extern crate semver;
 extern crate reqwest;
+extern crate futures;
+
+use reqwest::header::{Authorization, Bearer};
 
 use failure::{Error, ResultExt, SyncFailure};
 
@@ -33,10 +36,14 @@ use std::path::{Path, PathBuf};
 use std::io::{stderr, Write};
 use std::collections::BTreeMap;
 use std::sync::Arc;
+use std::thread::{self, JoinHandle};
 
 use chrono::{Timelike, Utc};
+use futures::sync::mpsc::{unbounded as unbounded_channel, UnboundedSender, UnboundedReceiver};
+use futures::stream::Stream;
 
 use collector::{Commit, ArtifactData, CommitData, Date};
+use collector::api::collected;
 use rust_sysroot::git::Commit as GitCommit;
 use rust_sysroot::sysroot::Sysroot;
 
@@ -144,6 +151,52 @@ fn kinds_from_arg<K>(strings_and_kinds: &[(&str, K)], arg: &str)
     Ok(v)
 }
 
+lazy_static! {
+    static ref BG_THREAD: Arc<(UnboundedSender<Option<collected::Request>>, JoinHandle<()>)> = start_bg_thread();
+}
+
+fn start_bg_thread() -> Arc<(UnboundedSender<Option<collected::Request>>, JoinHandle<()>)> {
+    let (sender, receiver): (_, UnboundedReceiver<Option<collected::Request>>) = unbounded_channel();
+    let sender_thread = sender.clone();
+    let handle = thread::spawn(move || {
+        for request in receiver.wait() {
+            let request = request.expect("result is Ok");
+            if request.is_none() {
+                // termination requested
+                break;
+            }
+            let request = request.unwrap();
+            let ret = (|| {
+                let client = reqwest::ClientBuilder::new().build()?;
+                let resp = client.post(&format!(
+                        "{}/perf/collected",
+                        env::var("SITE_URL").expect("SITE_URL defined")
+                    ))
+                    .header(Authorization(Bearer { token: env::var("PERF_SECRET_KEY").unwrap() }))
+                    .json(&request)
+                    .send()?;
+                if !resp.status().is_success() {
+                    bail!("ping home failed: {:?}", resp);
+                }
+                Ok(())
+            })();
+            match ret {
+                Ok(()) => {},
+                Err(err) => {
+                    warn!("call home failed: {:?}", err);
+                    thread::sleep(std::time::Duration::from_secs(10));
+                    sender_thread.unbounded_send(Some(request)).expect("can send on thread channel");
+                }
+            }
+        }
+    });
+    Arc::new((sender, handle))
+}
+
+fn send_home(d: &collected::Request) {
+    BG_THREAD.0.unbounded_send(Some(d.clone())).unwrap();
+}
+
 fn bench_commit(
     repo: Option<&outrepo::Repo>,
     commit: &GitCommit,
@@ -154,18 +207,37 @@ fn bench_commit(
     cargo_path: &Path,
     benchmarks: &[Benchmark],
     iterations: usize,
+    call_home: bool,
 ) -> CommitData {
     info!(
         "benchmarking commit {} ({}) for triple {}",
         commit.sha, commit.date, triple
     );
 
+    if call_home {
+        send_home(&collected::Request::BenchmarkCommit {
+            commit: Commit {
+                sha: commit.sha.clone(),
+                date: Date(commit.date),
+            },
+            benchmarks: benchmarks.iter().map(|b| b.name.clone()).collect(),
+        });
+    }
     let existing_data = repo.and_then(|r| r.load_commit_data(&commit, &triple).ok());
 
     let mut results = BTreeMap::new();
     if let Some(ref data) = existing_data {
         for benchmark in benchmarks {
             if let Some(result) = data.benchmarks.get(&benchmark.name) {
+                if call_home {
+                    send_home(&collected::Request::BenchmarkDone {
+                        benchmark: benchmark.name.clone(),
+                        commit: Commit {
+                            sha: commit.sha.clone(),
+                            date: Date(commit.date),
+                        },
+                    });
+                }
                 results.insert(benchmark.name.clone(), result.clone());
             }
         }
@@ -187,6 +259,16 @@ fn bench_commit(
             }
         };
 
+        if call_home {
+            send_home(&collected::Request::BenchmarkDone {
+                benchmark: benchmark.name.clone(),
+                commit: Commit {
+                    sha: commit.sha.clone(),
+                    date: Date(commit.date),
+                },
+            });
+        }
+
         results.insert(benchmark.name.clone(), result);
         info!("{} benchmarks left", benchmarks.len() - results.len());
     }
@@ -332,7 +414,7 @@ fn main_result() -> Result<i32, Error> {
         rust_sysroot::get_commits(rust_sysroot::EPOCH_COMMIT, "master").map_err(SyncFailure::new)
     };
 
-    match matches.subcommand() {
+    let ret = match matches.subcommand() {
         ("bench_commit", Some(sub_m)) => {
             let commit = sub_m.value_of("COMMIT").unwrap();
             let commit = get_commits()?
@@ -362,6 +444,7 @@ fn main_result() -> Result<i32, Error> {
                 &sysroot.cargo,
                 &benchmarks,
                 3,
+                false,
             ))?;
             Ok(0)
         }
@@ -398,6 +481,7 @@ fn main_result() -> Result<i32, Error> {
                 &cargo_path,
                 &benchmarks,
                 1,
+                false,
             );
             get_out_repo(true)?.add_commit_data(&result)?;
             Ok(0)
@@ -437,6 +521,7 @@ fn main_result() -> Result<i32, Error> {
                 &toolchain.binary_file("cargo"),
                 &benchmarks,
                 3,
+                false,
             );
             repo.success_artifact(&ArtifactData { id: id.to_string(), benchmarks: benchmark_data })?;
             Ok(0)
@@ -473,6 +558,7 @@ fn main_result() -> Result<i32, Error> {
                     &sysroot.cargo,
                     &benchmarks,
                     3,
+                    true,
                 ));
                 if let Err(err) = result {
                     out_repo.write_broken_commit(&commit, err)?;
@@ -556,6 +642,7 @@ fn main_result() -> Result<i32, Error> {
                     &sysroot.cargo,
                     &benchmarks,
                     1,
+                    false,
                 );
             } else {
                 panic!("no commits");
@@ -567,5 +654,8 @@ fn main_result() -> Result<i32, Error> {
             let _ = writeln!(stderr(), "{}", matches.usage());
             Ok(2)
         }
-    }
+    };
+    BG_THREAD.0.unbounded_send(None).unwrap();
+    thread::sleep(std::time::Duration::from_secs(1));
+    ret
 }

collector/src/bin/rustc-perf-collector/main.rs

@@ -19,6 +19,8 @@ use chrono::{DateTime, Datelike, Duration, TimeZone, Utc};
 use chrono::naive::NaiveDate;
 use serde::{Deserialize, Serialize};
 
+pub mod api;
+
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct Commit {
     pub sha: String,

collector/src/lib.rs

@@ -190,6 +190,7 @@ pub mod nll_dashboard {
 
 pub mod status {
     use collector::Commit;
+    use load::CurrentState;
 
     #[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
     pub struct BenchmarkStatus {
@@ -198,10 +199,12 @@ pub mod status {
         pub error: Option<String>,
     }
 
-    #[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
+    #[derive(Debug, Clone, Serialize, Deserialize)]
     pub struct Response {
         pub last_commit: Commit,
         pub benchmarks: Vec<BenchmarkStatus>,
         pub missing: Vec<Commit>,
+        pub current: Option<CurrentState>,
     }
 }
+

site/src/api.rs

@@ -14,7 +14,8 @@ use std::io::Read;
 use std::env;
 
 use serde_json;
-use failure::Error;
+use antidote::Mutex;
+use failure::{ResultExt, Error};
 use failure::SyncFailure;
 use rust_sysroot;
 use rust_sysroot::git::Commit as GitCommit;
@@ -26,6 +27,40 @@ use collector::Date;
 
 pub use collector::{Commit, CommitData, ArtifactData, Patch, Run, Stat};
 
+#[derive(Clone, Deserialize, Serialize, Debug)]
+pub struct CurrentState {
+    pub commit: Commit,
+    pub benchmarks: Vec<String>,
+}
+
+#[derive(Clone, Deserialize, Serialize, Debug)]
+pub struct Persistent {
+    pub try_commits: Vec<String>,
+    pub current: Option<CurrentState>,
+}
+
+impl Persistent {
+    pub fn write(&self) -> Result<(), Error> {
+        let s = serde_json::to_string(self)?;
+        fs::write("persistent.json", &s).with_context(|_| format!("failed to write persistent DB"))?;
+        Ok(())
+    }
+
+    fn load() -> Persistent {
+        Persistent::load_().unwrap_or_else(|| Persistent {
+            try_commits: Vec::new(),
+            current: None,
+        })
+    }
+
+    fn load_() -> Option<Persistent> {
+        let s = fs::read_to_string("persistent.json").ok()?;
+        let persistent: Persistent = serde_json::from_str(&s).ok()?;
+
+        Some(persistent)
+    }
+}
+
 #[derive(Debug)]
 pub struct InputData {
     /// A set containing all crate names of the bootstrap kind.
@@ -43,9 +78,9 @@ pub struct InputData {
 
     pub artifact_data: BTreeMap<String, ArtifactData>,
 
-    pub try_commits: Vec<String>,
-
     pub commits: Vec<GitCommit>,
+
+    pub persistent: Mutex<Persistent>,
 }
 
 impl InputData {
@@ -56,10 +91,6 @@ impl InputData {
         let mut artifact_data = BTreeMap::new();
         let mut data = BTreeMap::new();
 
-        let try_commits = fs::read_to_string("try-commit-queue")
-            .expect("could not open try-commit-queue file")
-            .lines().map(|s| s.to_owned()).collect::<Vec<_>>();
-
         if !repo_loc.exists() {
             // If the repository doesn't yet exist, simplify clone it to the given location.
             info!(
@@ -135,13 +166,12 @@ impl InputData {
         info!("{} skipped files", skipped);
         info!("{} measured", data.len());
 
-        InputData::new(data, artifact_data, try_commits)
+        InputData::new(data, artifact_data)
     }
 
     pub fn new(
         data: BTreeMap<Commit, CommitData>,
         artifact_data: BTreeMap<String, ArtifactData>,
-        try_commits: Vec<String>,
     ) -> Result<InputData, Error> {
         let mut last_date = None;
         let mut crate_list = BTreeSet::new();
@@ -177,7 +207,7 @@ impl InputData {
             data: data,
             artifact_data,
             commits,
-            try_commits,
+            persistent: Mutex::new(Persistent::load()),
         })
     }
 
@@ -205,8 +235,9 @@ impl InputData {
             })
             .collect::<Vec<_>>();
         missing.reverse();
-        Ok(self.try_commits.iter()
+        Ok(self.persistent.lock().try_commits.iter()
             .map(|sha| Commit { sha: sha.clone(), date: Date::ymd_hms(2001, 01, 01, 0, 0, 0) })
+            .filter(|c| !have.contains_key(&c.sha)) // we may have not updated the try-commits file
             .chain(missing)
             .collect())
     }

site/src/load.rs

@@ -24,5 +24,7 @@ fn main() {
         .unwrap_or(2346);
     println!("Starting server with port={:?}", port);
 
-    server::start(data, port);
+    let key = env::var("PERF_SECRET_KEY").unwrap_or(String::from("QuiteSecret"));
+
+    server::start(data, port, key);
 }