Record duration of collector benchmarking

collector/src/main.rs

@@ -13,7 +13,7 @@ use std::io::{stderr, Write};
 use std::path::{Path, PathBuf};
 use std::process;
 use std::process::Command;
-use std::str;
+use std::{str, time::Instant};
 use tokio::runtime::Runtime;
 
 mod background_worker;
@@ -274,6 +274,7 @@ fn bench(
     }
     let interned_cid = rt.block_on(tx.conn().artifact_id(&cid));
 
+    let start = Instant::now();
     for (nth_benchmark, benchmark) in benchmarks.iter().enumerate() {
         rt.block_on(
             tx.conn()
@@ -315,6 +316,11 @@ fn bench(
             }
         }
     }
+    let end = start.elapsed();
+
+    eprintln!("collection took {:?}", end);
+
+    rt.block_on(tx.conn().record_duration(interned_cid, end));
 
     // Publish results now that we've finished fully with this commit.
     rt.block_on(tx.commit()).unwrap();

database/src/pool.rs

@@ -2,6 +2,7 @@ use crate::{ArtifactId, ArtifactIdNumber};
 use crate::{Cache, CollectionId, Index, Profile, QueryDatum, QueuedCommit};
 use hashbrown::HashMap;
 use std::sync::{Arc, Mutex};
+use std::time::Duration;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 
 pub mod both;
@@ -15,6 +16,10 @@ pub trait Connection: Send + Sync {
 
     async fn load_index(&mut self) -> Index;
 
+    /// This records the duration of a collection run, i.e., collecting all of
+    /// the statistics for a particular artifact.
+    async fn record_duration(&self, artifact: ArtifactIdNumber, duration: Duration);
+
     async fn collection_id(&self) -> CollectionId;
     async fn artifact_id(&self, artifact: &ArtifactId) -> ArtifactIdNumber;
     async fn record_benchmark(&self, krate: &str, supports_stable: bool);

database/src/pool/both.rs

@@ -186,6 +186,16 @@ where
             self.b.record_benchmark(krate, supports_stable)
         );
     }
+    async fn record_duration(
+        &self,
+        artifact: crate::ArtifactIdNumber,
+        duration: std::time::Duration,
+    ) {
+        join!(
+            self.a.record_duration(artifact, duration),
+            self.b.record_duration(artifact, duration)
+        );
+    }
 }
 
 #[async_trait::async_trait]
@@ -349,4 +359,14 @@ impl<'a> Connection for BothTransaction<'a> {
             self.b.conn_ref().record_benchmark(krate, supports_stable)
         );
     }
+    async fn record_duration(
+        &self,
+        artifact: crate::ArtifactIdNumber,
+        duration: std::time::Duration,
+    ) {
+        join!(
+            self.a.conn_ref().record_duration(artifact, duration),
+            self.b.conn_ref().record_duration(artifact, duration)
+        );
+    }
 }

database/src/pool/postgres.rs

@@ -152,6 +152,13 @@ static MIGRATIONS: &[&str] = &[
     r#"
     create unique index on pull_request_build (pr) where complete = false;
     "#,
+    r#"
+    create table artifact_collection_duration(
+        aid smallint primary key not null references artifact(id) on delete cascade on update cascade,
+        date_recorded timestamptz not null,
+        duration integer not null
+    );
+    "#,
 ];
 
 #[async_trait::async_trait]
@@ -222,6 +229,7 @@ pub struct CachedStatements {
     select_pstat_series: Statement,
     get_error: Statement,
     collection_id: Statement,
+    record_duration: Statement,
 }
 
 pub struct PostgresTransaction<'a> {
@@ -336,6 +344,13 @@ impl PostgresConnection {
                 insert_pstat_series: conn.prepare("insert into pstat_series (crate, profile, cache, statistic) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING RETURNING id").await.unwrap(),
                 select_pstat_series: conn.prepare("select id from pstat_series where crate = $1 and profile = $2 and cache = $3 and statistic = $4").await.unwrap(),
                 collection_id: conn.prepare("insert into collection DEFAULT VALUES returning id").await.unwrap(),
+                record_duration: conn.prepare("
+                    insert into artifact_collection_duration (
+                        aid,
+                        date_recorded,
+                        duration
+                    ) VALUES ($1, CURRENT_TIMESTAMP, $2)
+                ").await.unwrap(),
             }),
             conn,
         }
@@ -356,6 +371,17 @@ where
             conn: tx,
         })
     }
+
+    async fn record_duration(&self, artifact: ArtifactIdNumber, duration: Duration) {
+        self.conn()
+            .execute(
+                &self.statements().record_duration,
+                &[&(artifact.0 as i16), &(duration.as_secs() as i32)],
+            )
+            .await
+            .unwrap();
+    }
+
     async fn load_index(&mut self) -> Index {
         Index {
             commits: self

database/src/pool/sqlite.rs

@@ -133,6 +133,13 @@ static MIGRATIONS: &[&str] = &[
         requested timestamp without time zone
     );
     "#,
+    r#"
+    create table artifact_collection_duration(
+        aid integer primary key not null references artifact(id) on delete cascade on update cascade,
+        date_recorded timestamp without time zone not null,
+        duration integer not null
+    );
+    "#,
 ];
 
 #[async_trait::async_trait]
@@ -209,6 +216,16 @@ impl Connection for SqliteConnection {
         })
     }
 
+    async fn record_duration(&self, artifact: ArtifactIdNumber, duration: Duration) {
+        self.raw_ref()
+            .prepare_cached(
+                "insert into artifact_collection_duration (aid, date_recorded, duration) VALUES (?, strftime('%s','now'), ?)",
+            )
+            .unwrap()
+            .execute(params![artifact.0, duration.as_secs() as i64])
+            .unwrap();
+    }
+
     async fn load_index(&mut self) -> Index {
         let commits = self
             .raw()
@@ -386,7 +403,7 @@ impl Connection for SqliteConnection {
     async fn queue_pr(&self, pr: u32) {
         self.raw_ref()
             .prepare_cached(
-                "insert into pull_request_builds (pr, complete, requested) VALUES (?, 0, now)",
+                "insert into pull_request_builds (pr, complete, requested) VALUES (?, 0, strftime('%s','now'))",
             )
             .unwrap()
             .execute(params![pr])