[TEMP] Reduce cache lock contention

crates/autopilot/src/solvable_orders.rs

@@ -47,10 +47,18 @@ pub struct Metrics {
     auction_update: IntCounterVec,
 
     /// Time taken to update the solvable orders cache.
+    #[metric(buckets(
+        0.1, 0.25, 0.5, 0.75, 1.0, 1.25, 1.5, 1.75, 2.0, 2.25, 2.5, 2.75, 3.0, 3.5, 4, 5
+    ))]
     auction_update_total_time: Histogram,
 
     /// Time spent on auction update individual stage.
-    #[metric(labels("stage"))]
+    #[metric(
+        labels("stage"),
+        buckets(
+            0.01, 0.05, 0.1, 0.25, 0.5, 0.75, 1.0, 1.25, 1.5, 1.75, 2.0, 2.5, 3.0, 3.5, 4.0, 5.0
+        )
+    )]
     auction_update_stage_time: HistogramVec,
 
     /// Auction creations.

crates/shared/Cargo.toml

@@ -44,6 +44,7 @@ order-validation = { path = "../order-validation" }
 primitive-types = { workspace = true }
 prometheus = { workspace = true }
 prometheus-metric-storage = { workspace = true }
+rand = { workspace = true }
 rate-limit = { path = "../rate-limit" }
 reqwest = { workspace = true, features = ["cookies", "gzip", "json"] }
 rust_decimal = { version = "1.35.0", features = ["maths"] }

crates/shared/src/price_estimation/native_price_cache.rs

@@ -10,6 +10,7 @@ use {
     indexmap::IndexSet,
     primitive_types::H160,
     prometheus::{IntCounter, IntCounterVec, IntGauge},
+    rand::Rng,
     std::{
         collections::{hash_map::Entry, HashMap},
         sync::{Arc, Mutex, MutexGuard, Weak},
@@ -226,14 +227,18 @@ impl UpdateTask {
 
 impl CachingNativePriceEstimator {
     pub async fn initialize_cache(&self, prices: HashMap<H160, BigDecimal>) {
-        let now = Instant::now();
-        // Update the cache to half max age time, so it gets fetched sooner, since we
-        // don't know exactly how old the price was
-        let updated_at = now - (self.0.max_age / 2);
+        let mut rng = rand::thread_rng();
+        let now = std::time::Instant::now();
 
         let cache = prices
             .into_iter()
             .filter_map(|(token, price)| {
+                // Generate random `updated_at` timestamp
+                // to avoid spikes of expired prices.
+                let percent_expired = rng.gen_range(50..=90);
+                let age = self.0.max_age.as_secs() * 100 / (100 - percent_expired);
+                let updated_at = now - Duration::from_secs(age);
+
                 Some((
                     token,
                     CachedResult {
@@ -317,26 +322,31 @@ impl CachingNativePriceEstimator {
         tokens: &'a [H160],
         timeout: Duration,
     ) -> HashMap<H160, NativePriceEstimateResult> {
+        let mut prices = self.get_cached_prices(tokens);
         if timeout.is_zero() {
-            return self.get_cached_prices(tokens);
+            return prices;
         }
 
-        let mut collected_prices = HashMap::new();
+        let uncached_tokens: Vec<_> = tokens
+            .iter()
+            .filter(|t| !prices.contains_key(t))
+            .copied()
+            .collect();
         let price_stream = self
             .0
-            .estimate_prices_and_update_cache(tokens, self.0.max_age);
+            .estimate_prices_and_update_cache(&uncached_tokens, self.0.max_age);
 
         let _ = time::timeout(timeout, async {
             let mut price_stream = price_stream;
 
             while let Some((token, result)) = price_stream.next().await {
-                collected_prices.insert(token, result);
+                prices.insert(token, result);
             }
         })
         .await;
 
         // Return whatever was collected up to that point, regardless of the timeout
-        collected_prices
+        prices
     }
 }
 

crates/shared/src/price_estimation/trade_finder.rs

@@ -12,11 +12,11 @@ use {
         Query,
     },
     crate::{
-        request_sharing::RequestSharing,
+        request_sharing::{BoxRequestSharing, RequestSharing},
         trade_finding::{TradeError, TradeFinding},
     },
     anyhow::{anyhow, Result},
-    futures::future::{BoxFuture, FutureExt as _},
+    futures::future::FutureExt,
     rate_limit::RateLimiter,
     std::sync::Arc,
 };
@@ -26,7 +26,7 @@ use {
 #[derive(Clone)]
 pub struct TradeEstimator {
     inner: Arc<Inner>,
-    sharing: RequestSharing<Arc<Query>, BoxFuture<'static, Result<Estimate, PriceEstimationError>>>,
+    sharing: Arc<BoxRequestSharing<Arc<Query>, Result<Estimate, PriceEstimationError>>>,
     rate_limiter: Arc<RateLimiter>,
 }
 
@@ -48,7 +48,7 @@ impl TradeEstimator {
                 finder,
                 verifier: None,
             }),
-            sharing: RequestSharing::labelled(format!("estimator_{}", label)),
+            sharing: Arc::new(RequestSharing::labelled(format!("estimator_{}", label))),
             rate_limiter,
         }
     }
@@ -69,7 +69,11 @@ impl TradeEstimator {
             )
             .boxed()
         };
-        self.sharing.shared_or_else(query, fut).await
+        // Clone `sharing` to prevent `fut` from capturing a reference to `self`
+        // which apparently creates a strong reference cycle causing the future
+        // returned by `.shared_or_else()` to never get freed.
+        let sharing = Arc::clone(&self.sharing);
+        sharing.shared_or_else(query, fut).await
     }
 }