Emit warnings when locks are held for too long

lithp · lithp · commit e57f34081539 · 2021-10-21T14:17:44.000-07:00
- Add rw_read and rw_write macros which wrap access to locks
  - They emit warnings if lock acqusition takes over 100ms
  - They emit warnings if locks are held for over 100ms

- Use rw_read and rw_write macros in most places RwLock's are accessed

Using these macros I was able to immediately identify the cause of a
deadlock during boot.
diff --git a/ethportal-peertest/src/main.rs b/ethportal-peertest/src/main.rs
@@ -11,6 +11,7 @@ use trin_core::portalnet::{
     types::{PortalnetConfig, ProtocolKind},
     Enr, U256,
 };
+use trin_core::rw_read;
 use trin_core::utils::setup_overlay_db;
 
 #[tokio::main]
@@ -27,9 +28,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         let discovery = Arc::new(RwLock::new(Discovery::new(portal_config).unwrap()));
         discovery.write().await.start().await.unwrap();
 
-        let db = Arc::new(setup_overlay_db(
-            discovery.read().await.local_enr().node_id(),
-        ));
+        let db = Arc::new(setup_overlay_db(rw_read!(discovery).local_enr().node_id()));
 
         let overlay = Arc::new(
             OverlayProtocol::new(
diff --git a/src/main.rs b/src/main.rs
@@ -8,6 +8,8 @@ use tokio::sync::RwLock;
 use trin_core::jsonrpc::handlers::JsonRpcHandler;
 use trin_core::jsonrpc::types::PortalJsonRpcRequest;
 use trin_core::portalnet::events::PortalnetEvents;
+use trin_core::rw_read;
+use trin_core::rw_write;
 use trin_core::{
     cli::{TrinConfig, HISTORY_NETWORK, STATE_NETWORK},
     jsonrpc::service::launch_jsonrpc_server,
@@ -52,12 +54,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let discovery = Arc::new(RwLock::new(
         Discovery::new(portalnet_config.clone()).unwrap(),
     ));
-    discovery.write().await.start().await.unwrap();
+    rw_write!(discovery).start().await.unwrap();
 
     // Setup Overlay database
-    let db = Arc::new(setup_overlay_db(
-        discovery.read().await.local_enr().node_id(),
-    ));
+    let db = Arc::new(setup_overlay_db(rw_read!(discovery).local_enr().node_id()));
 
     debug!("Selected networks to spawn: {:?}", trin_config.networks);
     // Initialize state sub-network service and event handlers, if selected
diff --git a/trin-core/src/jsonrpc/handlers.rs b/trin-core/src/jsonrpc/handlers.rs
@@ -9,6 +9,8 @@ use tokio::sync::RwLock;
 use crate::jsonrpc::endpoints::{Discv5Endpoint, HistoryEndpoint, StateEndpoint, TrinEndpoint};
 use crate::jsonrpc::types::{HistoryJsonRpcRequest, PortalJsonRpcRequest, StateJsonRpcRequest};
 use crate::portalnet::discovery::Discovery;
+use crate::rw_read;
+use crate::rw_write;
 
 type Responder<T, E> = mpsc::UnboundedSender<Result<T, E>>;
 
@@ -25,9 +27,9 @@ impl JsonRpcHandler {
         while let Some(request) = self.portal_jsonrpc_rx.recv().await {
             let response: Value = match request.endpoint {
                 TrinEndpoint::Discv5Endpoint(endpoint) => match endpoint {
-                    Discv5Endpoint::NodeInfo => self.discovery.read().await.node_info(),
+                    Discv5Endpoint::NodeInfo => rw_read!(self.discovery).node_info(),
                     Discv5Endpoint::RoutingTableInfo => {
-                        self.discovery.write().await.routing_table_info()
+                        rw_write!(self.discovery).routing_table_info()
                     }
                 },
                 TrinEndpoint::HistoryEndpoint(endpoint) => {
diff --git a/trin-core/src/lib.rs b/trin-core/src/lib.rs
@@ -6,3 +6,137 @@ pub mod jsonrpc;
 pub mod portalnet;
 pub mod socket;
 pub mod utils;
+
+pub const ACQUIRE_TIMEOUT_MS: u64 = 100;
+pub const HOLD_TIMEOUT_MS: u64 = 100;
+
+pub struct TimedRwReadGuard<'a, T: ?Sized> {
+    pub inner: tokio::sync::RwLockReadGuard<'a, T>,
+    pub acquisition_line: u32,
+    pub acquisition_file: &'static str,
+    pub acquisition_time: std::time::Instant,
+    pub sleep_task: tokio::task::JoinHandle<()>,
+}
+
+async fn sleep_then_log(file: &'static str, line: u32) {
+    tokio::time::sleep(std::time::Duration::from_millis(HOLD_TIMEOUT_MS)).await;
+    log::warn!(
+        "[{}:{}] lock held for over {}ms, not yet released",
+        file,
+        line,
+        HOLD_TIMEOUT_MS.to_string()
+    );
+}
+
+impl<'a, T: ?Sized> TimedRwReadGuard<'a, T> {
+    pub async fn new(
+        inner: tokio::sync::RwLockReadGuard<'a, T>,
+        acquisition_line: u32,
+        acquisition_file: &'static str,
+    ) -> TimedRwReadGuard<'a, T> {
+        let now = std::time::Instant::now();
+        let move_line = acquisition_line;
+        let move_file = acquisition_file;
+        let handle = tokio::spawn(async move {
+            sleep_then_log(move_file, move_line).await;
+        });
+
+        TimedRwReadGuard {
+            inner,
+            acquisition_line,
+            acquisition_file,
+            acquisition_time: now,
+            sleep_task: handle,
+        }
+    }
+}
+
+impl<'a, T> std::ops::Deref for TimedRwReadGuard<'a, T> {
+    type Target = tokio::sync::RwLockReadGuard<'a, T>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl<'a, T: ?Sized> Drop for TimedRwReadGuard<'a, T> {
+    fn drop(&mut self) {
+        self.sleep_task.abort();
+        let held_for = self.acquisition_time.elapsed().as_millis();
+        if held_for > HOLD_TIMEOUT_MS.into() {
+            log::warn!(
+                "[{}:{}] lock held for too long: {}ms",
+                self.acquisition_file,
+                self.acquisition_line,
+                held_for,
+            )
+        }
+    }
+}
+
+#[macro_export]
+macro_rules! rw_read {
+    ($lock_name:expr) => {{
+        let acquire_timeout = std::time::Duration::from_millis($crate::ACQUIRE_TIMEOUT_MS);
+        let sleep = tokio::time::sleep(acquire_timeout);
+        tokio::pin!(sleep);
+
+        let mut did_log: bool = false;
+        let now = std::time::Instant::now();
+
+        loop {
+            tokio::select! {
+                _ = &mut sleep, if !did_log => {
+                    log::warn!(
+                        "[{}:{}] waiting more than {}ms to acquire lock, still waiting",
+                        std::file!(),
+                        std::line!(),
+                        $crate::ACQUIRE_TIMEOUT_MS,
+                    );
+                    did_log = true;
+                }
+                guard = $lock_name.read() => {
+                    if did_log {
+                        let wait_time = now.elapsed().as_millis();
+                        log::warn!(
+                            "[{}:{}] waited {}ms to acquire lock",
+                            std::file!(),
+                            std::line!(),
+                            wait_time,
+                        );
+                    }
+
+                    let wrapped = $crate::TimedRwReadGuard::new(
+                        guard, std::line!(), std::file!(),
+                    ).await;
+                    break wrapped;
+                }
+            }
+        }
+    }};
+}
+
+#[macro_export]
+macro_rules! rw_write {
+    ($lock_name:expr) => {{
+        let acquire_timeout = std::time::Duration::from_millis($crate::ACQUIRE_TIMEOUT_MS);
+        let sleep = tokio::time::sleep(acquire_timeout);
+        tokio::pin!(sleep);
+
+        loop {
+            tokio::select! {
+                _ = &mut sleep, if !sleep.is_elapsed() => {
+                    log::warn!(
+                        "[{}:{}] took more than {} ms to acquire lock",
+                        std::file!(),
+                        std::line!(),
+                        $crate::ACQUIRE_TIMEOUT_MS,
+                    );
+                }
+                guard = $lock_name.write() => {
+                    break guard;
+                }
+            }
+        }
+    }};
+}
diff --git a/trin-core/src/portalnet/events.rs b/trin-core/src/portalnet/events.rs
@@ -10,6 +10,7 @@ use super::{
     utp::{UtpListener, UTP_PROTOCOL},
 };
 use crate::cli::{HISTORY_NETWORK, STATE_NETWORK};
+use crate::rw_write;
 use std::collections::HashMap;
 use std::convert::TryInto;
 
@@ -27,9 +28,7 @@ impl PortalnetEvents {
         history_sender: Option<mpsc::UnboundedSender<TalkRequest>>,
         state_sender: Option<mpsc::UnboundedSender<TalkRequest>>,
     ) -> Self {
-        let protocol_receiver = discovery
-            .write()
-            .await
+        let protocol_receiver = rw_write!(discovery)
             .discv5
             .event_stream()
             .await
diff --git a/trin-core/src/portalnet/overlay.rs b/trin-core/src/portalnet/overlay.rs
@@ -1,3 +1,4 @@
+use crate::rw_read;
 use crate::utils::xor_two_values;
 
 use super::{
@@ -9,6 +10,7 @@ use super::{
     Enr, U256,
 };
 use crate::portalnet::types::CustomPayload;
+use crate::rw_write;
 use discv5::{
     enr::NodeId,
     kbucket::{Filter, KBucketsTable},
@@ -112,7 +114,7 @@ impl OverlayProtocol {
         data_radius: U256,
     ) -> Self {
         let kbuckets = Arc::new(RwLock::new(KBucketsTable::new(
-            discovery.read().await.local_enr().node_id().into(),
+            rw_read!(discovery).local_enr().node_id().into(),
             config.bucket_pending_timeout,
             config.max_incoming_per_bucket,
             config.table_filter,
@@ -140,7 +142,7 @@ impl OverlayProtocol {
         let response = match request {
             Request::Ping(Ping { .. }) => {
                 debug!("Got overlay ping request {:?}", request);
-                let enr_seq = self.discovery.read().await.local_enr().seq();
+                let enr_seq = rw_read!(self.discovery).local_enr().seq();
                 let payload = CustomPayload::new(self.data_radius().await, None);
                 Response::Pong(Pong {
                     enr_seq,
@@ -181,12 +183,12 @@ impl OverlayProtocol {
 
     /// Returns the local ENR of the node.
     pub async fn local_enr(&self) -> Enr {
-        self.discovery.read().await.discv5.local_enr()
+        rw_read!(self.discovery).discv5.local_enr()
     }
 
     // Returns the data radius of the node.
     pub async fn data_radius(&self) -> U256 {
-        self.data_radius.read().await.clone()
+        rw_read!(self.data_radius).clone()
     }
 
     /// Returns a vector of the ENRs of the closest nodes by the given log2 distances.
@@ -203,7 +205,7 @@ impl OverlayProtocol {
         }
 
         if !log2_distances.is_empty() {
-            let mut kbuckets = self.kbuckets.write().await;
+            let mut kbuckets = rw_write!(self.kbuckets);
             for node in kbuckets
                 .nodes_by_distances(&log2_distances, FIND_NODES_MAX_NODES)
                 .into_iter()
@@ -246,19 +248,15 @@ impl OverlayProtocol {
 
     /// Returns a vector of all ENR node IDs of nodes currently contained in the routing table.
     pub async fn table_entries_id(&self) -> Vec<NodeId> {
-        self.kbuckets
-            .write()
-            .await
+        rw_write!(self.kbuckets)
             .iter()
             .map(|entry| *entry.node.key.preimage())
             .collect()
     }
 
     /// Returns a vector of all the ENRs of nodes currently contained in the routing table.
     pub async fn table_entries_enr(&self) -> Vec<Enr> {
-        self.kbuckets
-            .write()
-            .await
+        rw_write!(self.kbuckets)
             .iter()
             .map(|entry| entry.node.value.enr().clone())
             .collect()
@@ -271,16 +269,14 @@ impl OverlayProtocol {
         protocol: ProtocolKind,
         payload: Option<Vec<u8>>,
     ) -> Result<Vec<u8>, SendPingError> {
-        let enr_seq = self.discovery.read().await.local_enr().seq();
+        let enr_seq = rw_read!(self.discovery).local_enr().seq();
+
         let payload = CustomPayload::new(data_radius, payload);
         let msg = Ping {
             enr_seq,
             payload: Some(payload),
         };
-        Ok(self
-            .discovery
-            .read()
-            .await
+        Ok(rw_read!(self.discovery)
             .send_talkreq(
                 enr,
                 protocol.to_string(),
@@ -296,9 +292,7 @@ impl OverlayProtocol {
         protocol: ProtocolKind,
     ) -> Result<Vec<u8>, RequestError> {
         let msg = FindNodes { distances };
-        self.discovery
-            .read()
-            .await
+        rw_read!(self.discovery)
             .send_talkreq(
                 enr,
                 protocol.to_string(),
@@ -314,9 +308,7 @@ impl OverlayProtocol {
         protocol: ProtocolKind,
     ) -> Result<Vec<u8>, RequestError> {
         let msg = FindContent { content_key };
-        self.discovery
-            .read()
-            .await
+        rw_read!(self.discovery)
             .send_talkreq(
                 enr,
                 protocol.to_string(),
diff --git a/trin-core/src/portalnet/utp.rs b/trin-core/src/portalnet/utp.rs
@@ -12,6 +12,8 @@ use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::sync::RwLock;
 
+use crate::rw_read;
+
 pub const UTP_PROTOCOL: &str = "utp";
 pub const HEADER_SIZE: usize = 20;
 pub const MAX_DISCV5_PACKET_SIZE: usize = 1280;
@@ -433,7 +435,7 @@ impl UtpListener {
                         }
                     }
                     Type::StSyn => {
-                        if let Some(enr) = self.discovery.read().await.discv5.find_enr(&node_id) {
+                        if let Some(enr) = rw_read!(self.discovery).discv5.find_enr(&node_id) {
                             // If neither of those cases happened handle this is a new request
                             let mut conn = UtpStream::init(Arc::clone(&self.discovery), enr);
                             conn.handle_packet(packet).await;
@@ -466,7 +468,7 @@ impl UtpListener {
 
     // I am honestly not sure if I should init this with Enr or NodeId since we could use both
     async fn connect(&mut self, connection_id: u16, node_id: NodeId) {
-        if let Some(enr) = self.discovery.read().await.discv5.find_enr(&node_id) {
+        if let Some(enr) = rw_read!(self.discovery).discv5.find_enr(&node_id) {
             let mut conn = UtpStream::init(Arc::clone(&self.discovery), enr);
             conn.make_connection(connection_id).await;
             self.utp_connections.insert(
@@ -585,10 +587,7 @@ impl UtpStream {
             // It doesn't waste cpu cycles well waiting.
             tokio::time::sleep(Duration::from_millis(100)).await;
         }
-        let talk_request_result = self
-            .discovery
-            .read()
-            .await
+        let talk_request_result = rw_read!(self.discovery)
             .send_talkreq(self.enr.clone(), UTP_PROTOCOL.to_string(), packet.0.clone())
             .await;
         debug!("uTP TalkRequest result: {:?}", talk_request_result);
diff --git a/trin-state/src/events.rs b/trin-state/src/events.rs
@@ -4,6 +4,7 @@ use log::{debug, error, warn};
 use std::sync::Arc;
 use tokio::sync::{mpsc::UnboundedReceiver, RwLock};
 use trin_core::portalnet::types::Message;
+use trin_core::rw_write;
 
 pub struct StateEvents {
     pub network: Arc<RwLock<StateNetwork>>,
@@ -15,10 +16,7 @@ impl StateEvents {
         while let Some(talk_request) = self.event_rx.recv().await {
             debug!("Got state request {:?}", talk_request);
 
-            let reply = match self
-                .network
-                .write()
-                .await
+            let reply = match rw_write!(self.network)
                 .overlay
                 .process_one_request(&talk_request)
                 .await
diff --git a/trin-state/src/jsonrpc.rs b/trin-state/src/jsonrpc.rs
diff --git a/trin-state/src/lib.rs b/trin-state/src/lib.rs
diff --git a/trin-state/src/network.rs b/trin-state/src/network.rs
