feat(cardano-chain-follower): add thread and mmap file stats

Signed-off-by: bkioshn <[email protected]>

Author: bkioshn

rust/cardano-chain-follower/Cargo.toml

@@ -31,7 +31,7 @@ tokio = { version = "1.42.0", features = [
 ] }
 tracing = "0.1.41"
 tracing-log = "0.2.0"
-dashmap = "6.1.0"
+dashmap = { version = "6.1.0", features = ["serde"] }
 url = "2.5.4"
 anyhow = "1.0.95"
 chrono = "0.4.39"
@@ -56,6 +56,7 @@ ureq = { version = "2.12.1", features = ["native-certs"] }
 http = "1.2.0"
 hickory-resolver = { version = "0.24.2", features = ["dns-over-rustls"] }
 moka = { version = "0.12.9", features = ["sync"] }
+cpu-time = "1.0.0"
 
 [dev-dependencies]
 tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }

rust/cardano-chain-follower/src/chain_sync.rs

@@ -416,9 +416,14 @@ async fn live_sync_backfill_and_purge(
     cfg: ChainSyncConfig, mut rx: mpsc::Receiver<MithrilUpdateMessage>,
     mut sync_ready: SyncReadyWaiter,
 ) {
+    /// Thread name for stats.
+    const THREAD_NAME: &str = "LiveSyncBackfillAndPurge";
+
+    stats::start_thread(cfg.chain, THREAD_NAME, true);
     // Wait for first Mithril Update advice, which triggers a BACKFILL of the Live Data.
     let Some(update) = rx.recv().await else {
         error!("Mithril Sync Failed, can not continue chain sync either.");
+        stats::stop_thread(cfg.chain, THREAD_NAME);
         return;
     };
 
@@ -430,12 +435,15 @@ async fn live_sync_backfill_and_purge(
     let live_chain_head: Point;
 
     loop {
+        stats::resume_thread(cfg.chain, THREAD_NAME);
         // We will re-attempt backfill, until its successful.
         // Backfill is atomic, it either fully works, or none of the live-chain is changed.
         debug!("Mithril Tip has advanced to: {update:?} : BACKFILL");
         while let Err(error) = live_sync_backfill(&cfg, &update).await {
             error!("Mithril Backfill Sync Failed: {}", error);
+            stats::pause_thread(cfg.chain, THREAD_NAME);
             sleep(Duration::from_secs(10)).await;
+            stats::resume_thread(cfg.chain, THREAD_NAME);
         }
 
         if let Some(head_point) = get_live_head_point(cfg.chain) {
@@ -462,8 +470,11 @@ async fn live_sync_backfill_and_purge(
     let mut update_sender = get_chain_update_tx_queue(cfg.chain).await;
 
     loop {
+        stats::resume_thread(cfg.chain, THREAD_NAME);
+
         let Some(update) = rx.recv().await else {
             error!("Mithril Sync Failed, can not continue chain sync either.");
+            stats::stop_thread(cfg.chain, THREAD_NAME);
             return;
         };
 
@@ -516,6 +527,10 @@ async fn live_sync_backfill_and_purge(
 ///
 /// This does not return, it is a background task.
 pub(crate) async fn chain_sync(cfg: ChainSyncConfig, rx: mpsc::Receiver<MithrilUpdateMessage>) {
+    /// Thread name for stats.
+    const THREAD_NAME: &str = "ChainSync";
+
+    stats::start_thread(cfg.chain, THREAD_NAME, true);
     debug!(
         "Chain Sync for: {} from {} : Starting",
         cfg.chain, cfg.relay_address,
@@ -537,9 +552,9 @@ pub(crate) async fn chain_sync(cfg: ChainSyncConfig, rx: mpsc::Receiver<MithrilU
     let mut fork_count: Fork = Fork::FIRST_LIVE;
 
     loop {
+        stats::resume_thread(cfg.chain, THREAD_NAME);
         // We never have a connection if we end up around the loop, so make a new one.
         let mut peer = persistent_reconnect(&cfg.relay_address, cfg.chain).await;
-
         match resync_live_tip(&mut peer, cfg.chain).await {
             Ok(tip) => debug!("Tip Resynchronized to {tip}"),
             Err(error) => {

rust/cardano-chain-follower/src/chain_sync_ready.rs

@@ -12,7 +12,7 @@ use tokio::{
 };
 use tracing::error;
 
-use crate::chain_update;
+use crate::{chain_update, stats};
 
 /// Data we hold related to sync being ready or not.
 struct SyncReady {
@@ -85,9 +85,13 @@ static SYNC_READY: LazyLock<DashMap<Network, RwLock<SyncReady>>> = LazyLock::new
 /// Write Lock the `SYNC_READY` lock for a network.
 /// When we are signaled to be ready, set it to true and release the lock.
 pub(crate) fn wait_for_sync_ready(chain: Network) -> SyncReadyWaiter {
+    /// Thread name for stats.
+    const THREAD_NAME: &str = "WaitForSyncReady";
+
     let (tx, rx) = oneshot::channel::<()>();
 
     tokio::spawn(async move {
+        stats::start_thread(chain, THREAD_NAME, true);
         // We are safe to use `expect` here because the SYNC_READY list is exhaustively
         // initialized. Its a Serious BUG if that not True, so panic is OK.
         #[allow(clippy::expect_used)]
@@ -101,7 +105,7 @@ pub(crate) fn wait_for_sync_ready(chain: Network) -> SyncReadyWaiter {
         if let Ok(()) = rx.await {
             status.ready = true;
         }
-
+        stats::stop_thread(chain, THREAD_NAME);
         // If the channel closes early, we can NEVER use the Blockchain data.
     });
 

rust/cardano-chain-follower/src/lib.rs

@@ -14,6 +14,7 @@ mod mithril_snapshot_data;
 mod mithril_snapshot_iterator;
 mod mithril_snapshot_sync;
 mod mithril_turbo_downloader;
+mod mmap_file;
 mod snapshot_id;
 mod stats;
 pub mod turbo_downloader;

rust/cardano-chain-follower/src/mithril_query.rs

@@ -2,25 +2,35 @@
 
 use std::path::Path;
 
-use cardano_blockchain_types::Point;
+use cardano_blockchain_types::{Network, Point};
 use pallas_hardano::storage::immutable::FallibleBlock;
 use tokio::task;
 
-use crate::error::{Error, Result};
+use crate::{
+    error::{Error, Result},
+    stats,
+};
 
 /// Synchronous Immutable block iterator.
 pub(crate) type ImmutableBlockIterator = Box<dyn Iterator<Item = FallibleBlock> + Send + Sync>;
 
 /// Get a mithril snapshot iterator.
 pub(crate) async fn make_mithril_iterator(
-    path: &Path, start: &Point,
+    path: &Path, start: &Point, chain: Network,
 ) -> Result<ImmutableBlockIterator> {
+    /// Thread name for stats.
+    const THREAD_NAME: &str = "MithrilIterator";
+
     let path = path.to_path_buf();
     let start = start.clone();
     // Initial input
     let res = task::spawn_blocking(move || {
-        pallas_hardano::storage::immutable::read_blocks_from_point(&path, start.clone().into())
-            .map_err(|error| Error::MithrilSnapshot(Some(error)))
+        stats::start_thread(chain, THREAD_NAME, false);
+        let result =
+            pallas_hardano::storage::immutable::read_blocks_from_point(&path, start.clone().into())
+                .map_err(|error| Error::MithrilSnapshot(Some(error)));
+        stats::stop_thread(chain, THREAD_NAME);
+        result
     })
     .await;
 

rust/cardano-chain-follower/src/mithril_snapshot_iterator.rs

@@ -15,6 +15,7 @@ use tracing_log::log;
 use crate::{
     error::{Error, Result},
     mithril_query::{make_mithril_iterator, ImmutableBlockIterator},
+    stats,
 };
 
 /// Search backwards by 60 slots (seconds) looking for a previous block.
@@ -73,7 +74,7 @@ impl MithrilSnapshotIterator {
         chain: Network, path: &Path, from: &Point, search_interval: u64,
     ) -> Option<MithrilSnapshotIterator> {
         let point = probe_point(from, search_interval);
-        let Ok(mut iterator) = make_mithril_iterator(path, &point).await else {
+        let Ok(mut iterator) = make_mithril_iterator(path, &point, chain).await else {
             return None;
         };
 
@@ -116,7 +117,7 @@ impl MithrilSnapshotIterator {
         let this = this?;
 
         // Remake the iterator, based on the new known point.
-        let Ok(iterator) = make_mithril_iterator(path, &this).await else {
+        let Ok(iterator) = make_mithril_iterator(path, &this, chain).await else {
             return None;
         };
 
@@ -176,7 +177,7 @@ impl MithrilSnapshotIterator {
 
         debug!("Actual Mithril Iterator Start: {}", from);
 
-        let iterator = make_mithril_iterator(path, from).await?;
+        let iterator = make_mithril_iterator(path, from, chain).await?;
 
         Ok(MithrilSnapshotIterator {
             inner: Arc::new(Mutex::new(MithrilSnapshotIteratorInner {
@@ -191,12 +192,18 @@ impl MithrilSnapshotIterator {
     /// Get the next block, in a way that is Async friendly.
     /// Returns the next block, or None if there are no more blocks.
     pub(crate) async fn next(&self) -> Option<MultiEraBlock> {
+        /// Thread name for stats.
+        const THREAD_NAME: &str = "MithrilSnapshotIterator::Next";
+
         let inner = self.inner.clone();
 
         let res = task::spawn_blocking(move || {
             #[allow(clippy::unwrap_used)] // Unwrap is safe here because the lock can't be poisoned.
             let mut inner_iterator = inner.lock().unwrap();
-            inner_iterator.next()
+            stats::start_thread(inner_iterator.chain, THREAD_NAME, false);
+            let next = inner_iterator.next();
+            stats::stop_thread(inner_iterator.chain, THREAD_NAME);
+            next
         })
         .await;
 

rust/cardano-chain-follower/src/mithril_snapshot_sync.rs

@@ -295,14 +295,20 @@ async fn get_mithril_snapshot_and_certificate(
 async fn validate_mithril_snapshot(
     chain: Network, certificate: &MithrilCertificate, path: &Path,
 ) -> bool {
+    /// Thread name for stats.
+    const THREAD_NAME: &str = "ValidateMithrilSnapshot";
+
     let cert = certificate.clone();
     let mithril_path = path.to_path_buf();
     match tokio::spawn(async move {
         // This can be long running and CPU Intensive.
         // So we spawn it off to a background task.
-        MessageBuilder::new()
+        stats::start_thread(chain, THREAD_NAME, true);
+        let result = MessageBuilder::new()
             .compute_snapshot_message(&cert, &mithril_path)
-            .await
+            .await;
+        stats::stop_thread(chain, THREAD_NAME);
+        result
     })
     .await
     {
@@ -683,6 +689,10 @@ macro_rules! next_iteration {
 pub(crate) async fn background_mithril_update(
     cfg: MithrilSnapshotConfig, tx: Sender<MithrilUpdateMessage>,
 ) {
+    /// Thread name for stats.
+    const THREAD_NAME: &str = "MithrilSnapshotUpdater";
+
+    stats::start_thread(cfg.chain, THREAD_NAME, true);
     debug!(
         "Mithril Snapshot background updater for: {} from {} to {} : Starting",
         cfg.chain,
@@ -694,6 +704,7 @@ pub(crate) async fn background_mithril_update(
     let mut current_snapshot = recover_existing_snapshot(&cfg, &tx).await;
 
     loop {
+        stats::resume_thread(cfg.chain, THREAD_NAME);
         debug!("Background Mithril Updater - New Loop");
 
         cleanup(&cfg).await;