WIP: figure out overhead of linear histogram

Cargo.lock

@@ -2933,6 +2933,7 @@ dependencies = [
  "hdrhistogram",
  "humantime",
  "humantime-serde",
+ "once_cell",
  "pageserver",
  "rand 0.8.5",
  "serde",

pageserver/pagebench/Cargo.toml

@@ -11,6 +11,7 @@ clap.workspace = true
 hdrhistogram.workspace = true
 humantime.workspace = true
 humantime-serde.workspace = true
+once_cell.workspace = true
 rand.workspace = true
 serde.workspace = true
 serde_json.workspace = true

pageserver/pagebench/src/getpage_latest_lsn.rs

@@ -13,6 +13,8 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
+use crate::linear_histo::LinearHisto;
+
 /// Measure performance of the GetPage API, targeting the latest LSN.
 #[derive(clap::Parser)]
 pub(crate) struct Args {
@@ -81,7 +83,8 @@ struct PerTaskOutput {
 }
 
 struct PerTaskStats {
-    latency_histo: hdrhistogram::Histogram<u64>,
+    // latency_histo: hdrhistogram::Histogram<u64>,
+    linear_hist: LinearHisto,
 }
 
 impl PerTaskStats {
@@ -89,29 +92,30 @@ impl PerTaskStats {
         Self {
             // Initialize with fixed bounds so that we panic at runtime instead of resizing the histogram,
             // which would skew the benchmark results.
-            latency_histo: hdrhistogram::Histogram::new_with_bounds(1, 1_000_000_000, 3).unwrap(),
+            // latency_histo: hdrhistogram::Histogram::new_with_bounds(1, 1_000_000_000, 3).unwrap(),
+            linear_hist: LinearHisto::new(),
         }
     }
     fn observe(&mut self, latency: Duration) -> anyhow::Result<()> {
-        let micros: u64 = latency
-            .as_micros()
-            .try_into()
-            .context("latency greater than u64")?;
-        self.latency_histo
-            .record(micros)
-            .context("add to histogram")?;
+        // let micros: u64 = latency
+        //     .as_micros()
+        //     .try_into()
+        //     .context("latency greater than u64")?;
+        // // self.latency_histo
+        //     .record(micros)
+        //     .context("add to histogram")?;
+        self.linear_hist.observe(latency);
         Ok(())
     }
     fn output(&self) -> PerTaskOutput {
-        let latency_percentiles = std::array::from_fn(|idx| {
-            let micros = self
-                .latency_histo
-                .value_at_percentile(LATENCY_PERCENTILES[idx]);
-            Duration::from_micros(micros)
-        });
+        let latency_percentiles =
+            std::array::from_fn(|idx| self.linear_hist.percentile(LATENCY_PERCENTILES[idx]));
         PerTaskOutput {
-            request_count: self.latency_histo.len(),
-            latency_mean: Duration::from_micros(self.latency_histo.mean() as u64),
+            request_count: 0,
+            latency_mean: self
+                .linear_hist
+                .mean()
+                .unwrap_or_else(|| { eprintln!("{:?}", self.linear_hist); Duration::from_micros(666) }),
             latency_percentiles: LatencyPercentiles {
                 latency_percentiles,
             },
@@ -120,9 +124,11 @@ impl PerTaskStats {
 
     fn add(&mut self, other: &Self) {
         let Self {
-            ref mut latency_histo,
+            // ref mut latency_histo,
+            ref mut linear_hist,
         } = self;
-        latency_histo.add(&other.latency_histo).unwrap();
+        // latency_histo.add(&other.latency_histo).unwrap();
+        linear_hist.add(&other.linear_hist);
     }
 }
 

pageserver/pagebench/src/linear_histo.rs

@@ -0,0 +1,77 @@
+use std::time::Duration;
+
+
+const LATENCY_BUCKET_BASE: u64 = 0;
+const LATENCY_BUCKET_STEP: u64 = 100;
+const LATENCY_BUCKETS_LEN: usize = 10;
+
+#[derive(Debug)]
+pub(crate) struct LinearHisto {
+    below: u64,
+    counters: [u64; LATENCY_BUCKETS_LEN],
+    above: u64,
+}
+
+impl LinearHisto {
+    pub(crate) fn new() -> Self {
+        LinearHisto {
+            below: 0,
+            counters: std::array::from_fn(|_| 0),
+            above: 0,
+        }
+    }
+
+    pub(crate) fn observe(&mut self, latency: Duration) {
+        let latency: u64 = latency.as_micros().try_into().unwrap();
+        let bucket = if latency < LATENCY_BUCKET_BASE {
+            &mut self.below
+        } else if latency
+            >= (LATENCY_BUCKET_BASE + (LATENCY_BUCKETS_LEN as u64) * LATENCY_BUCKET_STEP)
+        {
+            &mut self.above
+        } else {
+            &mut self.counters[((latency - LATENCY_BUCKET_BASE) / LATENCY_BUCKET_STEP) as usize]
+        };
+        *bucket += 1;
+    }
+
+    pub(crate) fn add(&mut self, other: &Self) {
+        let Self {
+            ref mut below,
+            ref mut counters,
+            ref mut above,
+        } = self;
+        *below += other.below;
+        for i in 0..counters.len() {
+            counters[i] += other.counters[i];
+        }
+        *above += other.above;
+    }
+
+    fn bucket_lower(&self, idx: usize) -> u64 {
+        let idx = idx as u64;
+        LATENCY_BUCKET_BASE + idx * LATENCY_BUCKET_STEP
+    }
+    fn bucket_upper(&self, idx: usize) -> u64 {
+        let idx = idx as u64;
+        LATENCY_BUCKET_BASE + (idx + 1) * LATENCY_BUCKET_STEP
+    }
+
+    pub(crate) fn mean(&self) -> Option<Duration> {
+        if self.below > 0 || self.above > 0 {
+            return None;
+        }
+        let mut sum = 0;
+        let mut count = 0;
+        for (bucket_idx, counter) in self.counters.iter().enumerate() {
+            let bucket_mean = self.bucket_lower(bucket_idx) + self.bucket_upper(bucket_idx) / 2;
+            sum += counter * bucket_mean;
+            count += counter;
+        }
+        Some(Duration::from_micros(sum / count))
+    }
+
+    pub(crate) fn percentile(&self, p: f64) -> Duration {
+        Duration::from_micros(0)
+    }
+}

pageserver/pagebench/src/main.rs

@@ -2,17 +2,24 @@ use clap::Parser;
 
 mod getpage_latest_lsn;
 
+mod linear_histo;
+
 /// Component-level performance test for pageserver.
 #[derive(clap::Parser)]
 enum Args {
     GetPageLatestLsn(getpage_latest_lsn::Args),
 }
 
-#[tokio::main]
-async fn main() {
+fn main() {
     let args = Args::parse();
-    match args {
-        Args::GetPageLatestLsn(args) => getpage_latest_lsn::main(args).await,
-    }
-    .unwrap()
+
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .max_blocking_threads(1)
+        .build()
+        .unwrap();
+    let jh = match args {
+        Args::GetPageLatestLsn(args) => rt.spawn(getpage_latest_lsn::main(args)),
+    };
+    rt.block_on(jh).unwrap().unwrap();
 }