Address tokio contention messages

core/src/subgraph/context/mod.rs

@@ -23,7 +23,9 @@ use graph::{
     slog::Logger,
 };
 use std::collections::HashMap;
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
+
+use graph::parking_lot::RwLock;
 use tokio::sync::mpsc;
 
 use self::instance::SubgraphInstance;
@@ -44,18 +46,18 @@ impl SubgraphKeepAlive {
     }
 
     pub fn remove(&self, deployment_id: &DeploymentId) {
-        self.alive_map.write().unwrap().remove(deployment_id);
+        self.alive_map.write().remove(deployment_id);
         self.sg_metrics.running_count.dec();
     }
     pub fn insert(&self, deployment_id: DeploymentId, guard: CancelGuard) {
-        let old = self.alive_map.write().unwrap().insert(deployment_id, guard);
+        let old = self.alive_map.write().insert(deployment_id, guard);
         if old.is_none() {
             self.sg_metrics.running_count.inc();
         }
     }
 
     pub fn contains(&self, deployment_id: &DeploymentId) -> bool {
-        self.alive_map.read().unwrap().contains_key(deployment_id)
+        self.alive_map.read().contains_key(deployment_id)
     }
 }
 

graph/src/components/metrics/registry.rs

@@ -1,5 +1,7 @@
 use std::collections::HashMap;
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
+
+use crate::parking_lot::RwLock;
 
 use prometheus::{labels, Histogram, IntCounterVec};
 use prometheus::{IntCounter, IntGauge};
@@ -109,14 +111,13 @@ impl MetricsRegistry {
         };
         let counters = CounterVec::new(opts, variable_labels)?;
         let id = counters.desc().first().unwrap().id;
-        let maybe_counter = self.global_counter_vecs.read().unwrap().get(&id).cloned();
+        let maybe_counter = self.global_counter_vecs.read().get(&id).cloned();
         if let Some(counters) = maybe_counter {
             Ok(counters)
         } else {
             self.register(name, Box::new(counters.clone()));
             self.global_counter_vecs
                 .write()
-                .unwrap()
                 .insert(id, counters.clone());
             Ok(counters)
         }
@@ -161,15 +162,12 @@ impl MetricsRegistry {
     ) -> Result<Counter, PrometheusError> {
         let counter = counter_with_labels(name, help, const_labels)?;
         let id = counter.desc().first().unwrap().id;
-        let maybe_counter = self.global_counters.read().unwrap().get(&id).cloned();
+        let maybe_counter = self.global_counters.read().get(&id).cloned();
         if let Some(counter) = maybe_counter {
             Ok(counter)
         } else {
             self.register(name, Box::new(counter.clone()));
-            self.global_counters
-                .write()
-                .unwrap()
-                .insert(id, counter.clone());
+            self.global_counters.write().insert(id, counter.clone());
             Ok(counter)
         }
     }
@@ -210,15 +208,12 @@ impl MetricsRegistry {
     ) -> Result<Gauge, PrometheusError> {
         let gauge = gauge_with_labels(name, help, const_labels)?;
         let id = gauge.desc().first().unwrap().id;
-        let maybe_gauge = self.global_gauges.read().unwrap().get(&id).cloned();
+        let maybe_gauge = self.global_gauges.read().get(&id).cloned();
         if let Some(gauge) = maybe_gauge {
             Ok(gauge)
         } else {
             self.register(name, Box::new(gauge.clone()));
-            self.global_gauges
-                .write()
-                .unwrap()
-                .insert(id, gauge.clone());
+            self.global_gauges.write().insert(id, gauge.clone());
             Ok(gauge)
         }
     }
@@ -232,15 +227,12 @@ impl MetricsRegistry {
         let opts = Opts::new(name, help);
         let gauges = GaugeVec::new(opts, variable_labels)?;
         let id = gauges.desc().first().unwrap().id;
-        let maybe_gauge = self.global_gauge_vecs.read().unwrap().get(&id).cloned();
+        let maybe_gauge = self.global_gauge_vecs.read().get(&id).cloned();
         if let Some(gauges) = maybe_gauge {
             Ok(gauges)
         } else {
             self.register(name, Box::new(gauges.clone()));
-            self.global_gauge_vecs
-                .write()
-                .unwrap()
-                .insert(id, gauges.clone());
+            self.global_gauge_vecs.write().insert(id, gauges.clone());
             Ok(gauges)
         }
     }
@@ -254,14 +246,13 @@ impl MetricsRegistry {
         let opts = HistogramOpts::new(name, help);
         let histograms = HistogramVec::new(opts, variable_labels)?;
         let id = histograms.desc().first().unwrap().id;
-        let maybe_histogram = self.global_histogram_vecs.read().unwrap().get(&id).cloned();
+        let maybe_histogram = self.global_histogram_vecs.read().get(&id).cloned();
         if let Some(histograms) = maybe_histogram {
             Ok(histograms)
         } else {
             self.register(name, Box::new(histograms.clone()));
             self.global_histogram_vecs
                 .write()
-                .unwrap()
                 .insert(id, histograms.clone());
             Ok(histograms)
         }

graph/src/components/store/mod.rs

@@ -25,9 +25,11 @@ use std::collections::{BTreeMap, BTreeSet, HashSet};
 use std::fmt;
 use std::fmt::Display;
 use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 use std::time::Duration;
 
+use crate::parking_lot::RwLock;
+
 use async_trait::async_trait;
 
 use crate::blockchain::{Block, BlockHash, BlockPtr};

graph/src/data/graphql/load_manager.rs

@@ -4,9 +4,11 @@ use prometheus::core::GenericCounter;
 use rand::{prelude::Rng, rng};
 use std::collections::{HashMap, HashSet};
 use std::iter::FromIterator;
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
+use crate::parking_lot::RwLock;
+
 use crate::components::metrics::{Counter, GaugeVec, MetricsRegistry};
 use crate::components::store::{DeploymentId, PoolWaitStats};
 use crate::data::graphql::shape_hash::shape_hash;
@@ -57,7 +59,7 @@ impl ShardEffort {
     }
 
     pub fn add(&self, shard: &str, qref: QueryRef, duration: Duration, gauge: &GaugeVec) {
-        let mut inner = self.inner.write().unwrap();
+        let mut inner = self.inner.write();
         inner.add(qref, duration);
         gauge
             .with_label_values(&[shard])
@@ -70,7 +72,7 @@ impl ShardEffort {
     /// data for the particular query, return `None` as the effort
     /// for the query
     pub fn current_effort(&self, qref: &QueryRef) -> (Option<Duration>, Duration) {
-        let inner = self.inner.read().unwrap();
+        let inner = self.inner.read();
         let total_effort = inner.total.duration();
         let query_effort = inner.effort.get(qref).map(|stats| stats.duration());
         (query_effort, total_effort)
@@ -381,7 +383,7 @@ impl LoadManager {
 
         let qref = QueryRef::new(deployment, shape_hash);
 
-        if self.jailed_queries.read().unwrap().contains(&qref) {
+        if self.jailed_queries.read().contains(&qref) {
             return if ENV_VARS.load_simulate {
                 Proceed
             } else {
@@ -426,7 +428,7 @@ impl LoadManager {
                 "query_effort_ms" => query_effort,
                 "total_effort_ms" => total_effort,
                 "ratio" => format!("{:.4}", query_effort/total_effort));
-                self.jailed_queries.write().unwrap().insert(qref);
+                self.jailed_queries.write().insert(qref);
                 return if ENV_VARS.load_simulate {
                     Proceed
                 } else {
@@ -457,15 +459,15 @@ impl LoadManager {
     }
 
     fn overloaded(&self, wait_stats: &PoolWaitStats) -> (bool, Duration) {
-        let store_avg = wait_stats.read().unwrap().average();
+        let store_avg = wait_stats.read().average();
         let overloaded = store_avg
             .map(|average| average > ENV_VARS.load_threshold)
             .unwrap_or(false);
         (overloaded, store_avg.unwrap_or(Duration::ZERO))
     }
 
     fn kill_state(&self, shard: &str) -> (f64, Instant) {
-        let state = self.kill_state.get(shard).unwrap().read().unwrap();
+        let state = self.kill_state.get(shard).unwrap().read();
         (state.kill_rate, state.last_update)
     }
 
@@ -505,7 +507,7 @@ impl LoadManager {
                 kill_rate = (kill_rate - KILL_RATE_STEP_DOWN).max(0.0);
             }
             let event = {
-                let mut state = self.kill_state.get(shard).unwrap().write().unwrap();
+                let mut state = self.kill_state.get(shard).unwrap().write();
                 state.kill_rate = kill_rate;
                 state.last_update = now;
                 state.log_event(now, kill_rate, overloaded)

graph/src/env/store.rs

@@ -163,6 +163,9 @@ pub struct EnvVarsStore {
     /// Disables storing or reading `eth_call` results from the store call cache.
     /// Set by `GRAPH_STORE_DISABLE_CALL_CACHE`. Defaults to false.
     pub disable_call_cache: bool,
+    /// Set by `GRAPH_STORE_DISABLE_CHAIN_HEAD_PTR_CACHE`. Default is false.
+    /// Set to true to disable chain_head_ptr caching (safety escape hatch).
+    pub disable_chain_head_ptr_cache: bool,
 }
 
 // This does not print any values avoid accidentally leaking any sensitive env vars
@@ -224,6 +227,7 @@ impl TryFrom<InnerStore> for EnvVarsStore {
             account_like_min_versions_count: x.account_like_min_versions_count,
             account_like_max_unique_ratio: x.account_like_max_unique_ratio.map(|r| r.0),
             disable_call_cache: x.disable_call_cache,
+            disable_chain_head_ptr_cache: x.disable_chain_head_ptr_cache,
         };
         if let Some(timeout) = vars.batch_timeout {
             if timeout < 2 * vars.batch_target_duration {
@@ -331,6 +335,8 @@ pub struct InnerStore {
     account_like_max_unique_ratio: Option<ZeroToOneF64>,
     #[envconfig(from = "GRAPH_STORE_DISABLE_CALL_CACHE", default = "false")]
     disable_call_cache: bool,
+    #[envconfig(from = "GRAPH_STORE_DISABLE_CHAIN_HEAD_PTR_CACHE", default = "false")]
+    disable_chain_head_ptr_cache: bool,
 }
 
 #[derive(Clone, Copy, Debug)]

graph/src/util/timed_cache.rs

@@ -3,10 +3,12 @@ use std::{
     cmp::Eq,
     collections::HashMap,
     hash::Hash,
-    sync::{Arc, RwLock},
+    sync::Arc,
     time::{Duration, Instant},
 };
 
+use crate::parking_lot::RwLock;
+
 /// Caching of values for a specified amount of time
 #[derive(Debug)]
 struct CacheEntry<V> {
@@ -49,7 +51,7 @@ impl<K, V> TimedCache<K, V> {
         K: Borrow<Q> + Eq + Hash,
         Q: Hash + Eq + ?Sized,
     {
-        match self.entries.read().unwrap().get(key) {
+        match self.entries.read().get(key) {
             Some(CacheEntry { value, expires }) if expires >= &now => Some(value.clone()),
             _ => None,
         }
@@ -72,11 +74,11 @@ impl<K, V> TimedCache<K, V> {
             value,
             expires: now + self.ttl,
         };
-        self.entries.write().unwrap().insert(key, entry);
+        self.entries.write().insert(key, entry);
     }
 
     pub fn clear(&self) {
-        self.entries.write().unwrap().clear();
+        self.entries.write().clear();
     }
 
     pub fn find<F>(&self, pred: F) -> Option<Arc<V>>
@@ -85,7 +87,6 @@ impl<K, V> TimedCache<K, V> {
     {
         self.entries
             .read()
-            .unwrap()
             .values()
             .find(move |entry| pred(entry.value.as_ref()))
             .map(|entry| entry.value.clone())
@@ -101,7 +102,6 @@ impl<K, V> TimedCache<K, V> {
     {
         self.entries
             .write()
-            .unwrap()
             .remove(key)
             .map(|CacheEntry { value, expires }| (value, expires >= Instant::now()))
     }

store/postgres/src/block_store.rs

@@ -1,8 +1,6 @@
-use std::{
-    collections::HashMap,
-    sync::{Arc, RwLock},
-    time::Duration,
-};
+use std::{collections::HashMap, sync::Arc, time::Duration};
+
+use graph::parking_lot::RwLock;
 
 use anyhow::anyhow;
 use async_trait::async_trait;
@@ -321,7 +319,6 @@ impl BlockStore {
         let configured_chains = block_store
             .stores
             .read()
-            .unwrap()
             .keys()
             .cloned()
             .collect::<Vec<_>>();
@@ -410,7 +407,6 @@ impl BlockStore {
         let store = Arc::new(store);
         self.stores
             .write()
-            .unwrap()
             .insert(chain.name.clone(), store.clone());
         Ok(store)
     }
@@ -475,12 +471,7 @@ impl BlockStore {
     }
 
     async fn store(&self, chain: &str) -> Option<Arc<ChainStore>> {
-        let store = self
-            .stores
-            .read()
-            .unwrap()
-            .get(chain)
-            .map(CheapClone::cheap_clone);
+        let store = self.stores.read().get(chain).map(CheapClone::cheap_clone);
         if store.is_some() {
             return store;
         }
@@ -506,7 +497,7 @@ impl BlockStore {
 
         chain_store.drop_chain().await?;
 
-        self.stores.write().unwrap().remove(chain);
+        self.stores.write().remove(chain);
 
         Ok(())
     }
@@ -516,7 +507,6 @@ impl BlockStore {
     fn stores(&self) -> Vec<Arc<ChainStore>> {
         self.stores
             .read()
-            .unwrap()
             .values()
             .map(CheapClone::cheap_clone)
             .collect()