graph, store: Add lock-free AtomicMovingStats for pool wait tracking

Replace RwLock<MovingStats> with a lock-free AtomicMovingStats that uses
an atomic ring buffer with packed bins. Each bin packs epoch (32 bits),
count (32 bits), and duration_nanos (64 bits) into a single AtomicU128
for lock-free CAS updates.

This eliminates lock contention when many threads write concurrently
(every semaphore wait, connection checkout, query execution) while
reducing memory usage by 2x (4.8KB vs 9.6KB per stats instance).

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: lutter

Cargo.lock

@@ -79,6 +79,7 @@ futures03 = { version = "0.3.31", package = "futures", features = ["compat"] }
 wasmparser = "0.118.1"
 thiserror = { workspace = true }
 parking_lot = "0.12.5"
+portable-atomic = { version = "1.11", features = ["fallback"] }
 itertools = "0.14.0"
 defer = "0.2"
 

graph/Cargo.toml

@@ -28,8 +28,6 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
-use crate::parking_lot::RwLock;
-
 use async_trait::async_trait;
 
 use crate::blockchain::{Block, BlockHash, BlockPtr};
@@ -44,7 +42,7 @@ use crate::env::ENV_VARS;
 use crate::internal_error;
 use crate::prelude::{s, Attribute, DeploymentHash, ValueType};
 use crate::schema::{ast as sast, EntityKey, EntityType, InputSchema};
-use crate::util::stats::MovingStats;
+use crate::util::stats::AtomicMovingStats;
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct EntityFilterDerivative(bool);
@@ -744,8 +742,8 @@ impl Display for DeploymentLocator {
 }
 
 // The type that the connection pool uses to track wait times for
-// connection checkouts
-pub type PoolWaitStats = Arc<RwLock<MovingStats>>;
+// connection checkouts. Uses lock-free atomic operations internally.
+pub type PoolWaitStats = Arc<AtomicMovingStats>;
 
 /// Determines which columns should be selected in a table.
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]

graph/src/components/store/mod.rs

@@ -459,7 +459,7 @@ impl LoadManager {
     }
 
     fn overloaded(&self, wait_stats: &PoolWaitStats) -> (bool, Duration) {
-        let store_avg = wait_stats.read().average();
+        let store_avg = wait_stats.average();
         let overloaded = store_avg
             .map(|average| average > ENV_VARS.load_threshold)
             .unwrap_or(false);

graph/src/data/graphql/load_manager.rs

@@ -170,7 +170,7 @@ pub mod prelude {
     pub use crate::log::split::split_logger;
     pub use crate::util::cache_weight::CacheWeight;
     pub use crate::util::futures::{retry, TimeoutError};
-    pub use crate::util::stats::MovingStats;
+    pub use crate::util::stats::{AtomicMovingStats, MovingStats};
 
     macro_rules! static_graphql {
         ($m:ident, $m2:ident, {$($n:ident,)*}) => {

graph/src/lib.rs

@@ -1,6 +1,8 @@
 use std::collections::VecDeque;
+use std::sync::atomic::Ordering;
 use std::time::{Duration, Instant};
 
+use portable_atomic::AtomicU128;
 use prometheus::Gauge;
 
 use crate::prelude::ENV_VARS;
@@ -166,6 +168,196 @@ impl MovingStats {
     }
 }
 
+/// Packed bin for atomic operations: epoch (32 bits) | count (32 bits) | duration_nanos (64 bits)
+/// Fits in a single AtomicU128 for lock-free CAS updates.
+#[repr(transparent)]
+struct PackedBin(AtomicU128);
+
+impl PackedBin {
+    fn new() -> Self {
+        Self(AtomicU128::new(0))
+    }
+
+    /// Pack epoch, count, and duration into a single u128 value.
+    fn pack(epoch: u32, count: u32, duration_nanos: u64) -> u128 {
+        ((epoch as u128) << 96) | ((count as u128) << 64) | (duration_nanos as u128)
+    }
+
+    /// Unpack a u128 value into (epoch, count, duration_nanos).
+    fn unpack(packed: u128) -> (u32, u32, u64) {
+        let epoch = (packed >> 96) as u32;
+        let count = (packed >> 64) as u32;
+        let duration_nanos = packed as u64;
+        (epoch, count, duration_nanos)
+    }
+}
+
+/// Lock-free moving statistics using an epoch-based ring buffer.
+///
+/// This is a thread-safe, lock-free alternative to `MovingStats` that uses
+/// atomic operations instead of locks. It tracks durations over a sliding
+/// time window, storing values in fixed-size bins.
+///
+/// Writers use CAS loops to atomically update bins, while readers can
+/// scan all bins without blocking writers.
+pub struct AtomicMovingStats {
+    start_time: Instant,
+    bin_size: Duration,
+    bins: Box<[PackedBin]>,
+}
+
+impl Default for AtomicMovingStats {
+    fn default() -> Self {
+        Self::new(ENV_VARS.load_window_size, ENV_VARS.load_bin_size)
+    }
+}
+
+impl AtomicMovingStats {
+    /// Create a new AtomicMovingStats with the given window and bin sizes.
+    ///
+    /// # Panics
+    ///
+    /// Panics if `window_size` or `bin_size` is `0`, or if `bin_size` >= `window_size`
+    pub fn new(window_size: Duration, bin_size: Duration) -> Self {
+        assert!(window_size.as_millis() > 0);
+        assert!(bin_size.as_millis() > 0);
+        assert!(window_size > bin_size);
+
+        let num_bins = window_size.as_millis() as usize / bin_size.as_millis() as usize;
+        let bins: Vec<PackedBin> = (0..num_bins).map(|_| PackedBin::new()).collect();
+
+        Self {
+            start_time: Instant::now(),
+            bin_size,
+            bins: bins.into_boxed_slice(),
+        }
+    }
+
+    /// Calculate the epoch number for a given instant.
+    fn epoch_at(&self, now: Instant) -> u32 {
+        let elapsed = now.saturating_duration_since(self.start_time);
+        (elapsed.as_millis() / self.bin_size.as_millis()) as u32
+    }
+
+    /// Add a duration measurement at the current time.
+    pub fn add(&self, duration: Duration) {
+        self.add_at(Instant::now(), duration);
+    }
+
+    /// Add a duration measurement at a specific time.
+    ///
+    /// Note: It is expected that subsequent calls to `add_at` happen with
+    /// monotonically increasing `now` values for optimal accuracy.
+    pub fn add_at(&self, now: Instant, duration: Duration) {
+        let current_epoch = self.epoch_at(now);
+        let bin_idx = current_epoch as usize % self.bins.len();
+        let bin = &self.bins[bin_idx];
+        let duration_nanos = duration.as_nanos() as u64;
+
+        loop {
+            let current = bin.0.load(Ordering::Acquire);
+            let (bin_epoch, count, total_nanos) = PackedBin::unpack(current);
+
+            let new_packed = if bin_epoch == current_epoch {
+                // Same epoch: increment existing values
+                PackedBin::pack(
+                    current_epoch,
+                    count.saturating_add(1),
+                    total_nanos.saturating_add(duration_nanos),
+                )
+            } else {
+                // Stale epoch: reset bin with new measurement
+                PackedBin::pack(current_epoch, 1, duration_nanos)
+            };
+
+            match bin.0.compare_exchange_weak(
+                current,
+                new_packed,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            ) {
+                Ok(_) => break,
+                Err(_) => continue, // CAS failed, retry
+            }
+        }
+    }
+
+    /// Calculate the average duration over the current window.
+    ///
+    /// Returns `None` if no measurements have been recorded in the window.
+    pub fn average(&self) -> Option<Duration> {
+        self.average_at(Instant::now())
+    }
+
+    /// Calculate the average duration at a specific time.
+    fn average_at(&self, now: Instant) -> Option<Duration> {
+        let current_epoch = self.epoch_at(now);
+        let num_bins = self.bins.len() as u32;
+        let mut total_count: u64 = 0;
+        let mut total_nanos: u128 = 0;
+
+        for bin in self.bins.iter() {
+            let (bin_epoch, count, duration_nanos) =
+                PackedBin::unpack(bin.0.load(Ordering::Acquire));
+            // Valid if within window (handles epoch wraparound)
+            if current_epoch.wrapping_sub(bin_epoch) < num_bins {
+                total_count += count as u64;
+                total_nanos += duration_nanos as u128;
+            }
+        }
+
+        if total_count > 0 {
+            Some(Duration::from_nanos(
+                (total_nanos / total_count as u128) as u64,
+            ))
+        } else {
+            None
+        }
+    }
+
+    /// Return `true` if the average of measurements within the window
+    /// is above `duration`.
+    pub fn average_gt(&self, duration: Duration) -> bool {
+        self.average().map(|avg| avg > duration).unwrap_or(false)
+    }
+
+    /// Return `true` if the average at a specific time is above `duration`.
+    #[cfg(test)]
+    fn average_gt_at(&self, now: Instant, duration: Duration) -> bool {
+        self.average_at(now)
+            .map(|avg| avg > duration)
+            .unwrap_or(false)
+    }
+
+    /// Return the total duration recorded in the current window.
+    pub fn duration(&self) -> Duration {
+        self.duration_at(Instant::now())
+    }
+
+    /// Return the total duration at a specific time.
+    fn duration_at(&self, now: Instant) -> Duration {
+        let current_epoch = self.epoch_at(now);
+        let num_bins = self.bins.len() as u32;
+        let mut total_nanos: u128 = 0;
+
+        for bin in self.bins.iter() {
+            let (bin_epoch, _, duration_nanos) = PackedBin::unpack(bin.0.load(Ordering::Acquire));
+            if current_epoch.wrapping_sub(bin_epoch) < num_bins {
+                total_nanos += duration_nanos as u128;
+            }
+        }
+
+        Duration::from_nanos(total_nanos as u64)
+    }
+
+    /// Adds `duration` to the stats, and register the average ms to `avg_gauge`.
+    pub fn add_and_register(&self, duration: Duration, avg_gauge: &Gauge) {
+        self.add(duration);
+        let wait_avg = self.average().map(|avg| avg.as_millis()).unwrap_or(0);
+        avg_gauge.set(wait_avg as f64);
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -219,4 +411,77 @@ mod tests {
         assert_eq!(20, stats.total.count);
         assert_eq!(Duration::from_secs(5 * 86 + 16 * 10), stats.total.duration);
     }
+
+    #[test]
+    fn atomic_add_one_const() {
+        let stats = AtomicMovingStats::new(Duration::from_secs(5), Duration::from_secs(1));
+        let start = stats.start_time;
+        for i in 0..10 {
+            stats.add_at(start + Duration::from_secs(i), Duration::from_secs(1));
+        }
+        // After 10 seconds with 5-second window, only last 5 entries are valid
+        assert_eq!(5, stats.bins.len());
+        // Query at time 10 seconds (end of data range)
+        let query_time = start + Duration::from_secs(10);
+        // Average should be 1 second
+        let avg = stats.average_at(query_time).unwrap();
+        assert_eq!(Duration::from_secs(1), avg);
+        assert!(stats.average_gt_at(query_time, Duration::from_millis(900)));
+        assert!(!stats.average_gt_at(query_time, Duration::from_secs(1)));
+    }
+
+    #[test]
+    fn atomic_add_four_linear() {
+        let stats = AtomicMovingStats::new(Duration::from_secs(5), Duration::from_secs(1));
+        let start = stats.start_time;
+        for i in 0..40u64 {
+            stats.add_at(
+                start + Duration::from_millis(250 * i),
+                Duration::from_secs(i),
+            );
+        }
+        assert_eq!(5, stats.bins.len());
+        // Query at time 9.999 seconds (just before epoch 10 to include epoch 5)
+        // At epoch 9, valid bins are epochs 5-9 (9 - bin_epoch < 5)
+        let query_time = start + Duration::from_millis(9999);
+        // Total duration in window: 4 entries per bin, bins 5-9 contain entries 20-39
+        // Bin 5: entries 20,21,22,23 -> sum = 86
+        // Bin 6: entries 24,25,26,27 -> sum = 102
+        // ...
+        // Total count = 20, total duration = 5*86 + 16*10 = 590
+        assert_eq!(
+            Duration::from_secs(5 * 86 + 16 * 10),
+            stats.duration_at(query_time)
+        );
+    }
+
+    #[test]
+    fn atomic_empty_average() {
+        let stats = AtomicMovingStats::new(Duration::from_secs(5), Duration::from_secs(1));
+        // Query at start_time (no data yet)
+        assert!(stats.average_at(stats.start_time).is_none());
+        assert!(!stats.average_gt_at(stats.start_time, Duration::from_secs(1)));
+    }
+
+    #[test]
+    fn atomic_pack_unpack() {
+        // Test edge cases of packing/unpacking
+        let packed = PackedBin::pack(u32::MAX, u32::MAX, u64::MAX);
+        let (epoch, count, nanos) = PackedBin::unpack(packed);
+        assert_eq!(u32::MAX, epoch);
+        assert_eq!(u32::MAX, count);
+        assert_eq!(u64::MAX, nanos);
+
+        let packed = PackedBin::pack(0, 0, 0);
+        let (epoch, count, nanos) = PackedBin::unpack(packed);
+        assert_eq!(0, epoch);
+        assert_eq!(0, count);
+        assert_eq!(0, nanos);
+
+        let packed = PackedBin::pack(12345, 67890, 123456789012345);
+        let (epoch, count, nanos) = PackedBin::unpack(packed);
+        assert_eq!(12345, epoch);
+        assert_eq!(67890, count);
+        assert_eq!(123456789012345, nanos);
+    }
 }

graph/src/util/stats.rs

@@ -10,10 +10,10 @@ use diesel_async::pooled_connection::{PoolError as DieselPoolError, PoolableConn
 use diesel_async::{AsyncConnection, RunQueryDsl};
 use graph::env::ENV_VARS;
 use graph::prelude::error;
+use graph::prelude::AtomicMovingStats;
 use graph::prelude::Counter;
 use graph::prelude::Gauge;
 use graph::prelude::MetricsRegistry;
-use graph::prelude::MovingStats;
 use graph::prelude::PoolWaitStats;
 use graph::slog::info;
 use graph::slog::Logger;
@@ -24,8 +24,6 @@ use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use graph::parking_lot::RwLock;
-
 use crate::pool::AsyncPool;
 
 /// Our own connection manager. It is pretty much the same as
@@ -298,7 +296,7 @@ impl WaitMeter {
                 const_labels,
             )
             .expect("failed to create `store_connection_wait_time_ms` counter");
-        let wait_stats = Arc::new(RwLock::new(MovingStats::default()));
+        let wait_stats = Arc::new(AtomicMovingStats::default());
 
         Self {
             wait_gauge,
@@ -307,8 +305,6 @@ impl WaitMeter {
     }
 
     pub(crate) fn add_conn_wait_time(&self, duration: Duration) {
-        self.wait_stats
-            .write()
-            .add_and_register(duration, &self.wait_gauge);
+        self.wait_stats.add_and_register(duration, &self.wait_gauge);
     }
 }