fix(cardano-chain-follower): thread stat calling

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

Author: bkioshn

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

@@ -411,31 +411,14 @@ async fn live_sync_backfill(
     Ok(())
 }
 
-/// Call the live sync backfill.
-/// This is a helper function to pause and resume the stats thread.
-async fn call_live_sync_backfill(
-    cfg: &ChainSyncConfig, update: &MithrilUpdateMessage,
-) -> anyhow::Result<()> {
-    stats::pause_thread(cfg.chain, stats::thread::name::LIVE_SYNC_BACKFILL_AND_PURGE);
-    let result = live_sync_backfill(cfg, update).await;
-    stats::resume_thread(cfg.chain, stats::thread::name::LIVE_SYNC_BACKFILL_AND_PURGE);
-    result
-}
-
 /// Backfill and Purge the live chain, based on the Mithril Sync updates.
 async fn live_sync_backfill_and_purge(
     cfg: ChainSyncConfig, mut rx: mpsc::Receiver<MithrilUpdateMessage>,
     mut sync_ready: SyncReadyWaiter,
 ) {
-    stats::start_thread(
-        cfg.chain,
-        stats::thread::name::LIVE_SYNC_BACKFILL_AND_PURGE,
-        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, stats::thread::name::LIVE_SYNC_BACKFILL_AND_PURGE);
         return;
     };
 
@@ -450,7 +433,7 @@ async fn live_sync_backfill_and_purge(
         // 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) = call_live_sync_backfill(&cfg, &update).await {
+        while let Err(error) = live_sync_backfill(&cfg, &update).await {
             error!("Mithril Backfill Sync Failed: {}", error);
             sleep(Duration::from_secs(10)).await;
         }
@@ -481,7 +464,6 @@ async fn live_sync_backfill_and_purge(
     loop {
         let Some(update) = rx.recv().await else {
             error!("Mithril Sync Failed, can not continue chain sync either.");
-            stats::stop_thread(cfg.chain, stats::thread::name::LIVE_SYNC_BACKFILL_AND_PURGE);
             return;
         };
 
@@ -546,7 +528,13 @@ pub(crate) async fn chain_sync(cfg: ChainSyncConfig, rx: mpsc::Receiver<MithrilU
 
     // Start the Live chain backfill task.
     let _backfill_join_handle = spawn(async move {
+        stats::start_thread(
+            cfg.chain,
+            stats::thread::name::LIVE_SYNC_BACKFILL_AND_PURGE,
+            true,
+        );
         live_sync_backfill_and_purge(backfill_cfg.clone(), rx, sync_waiter).await;
+        stats::stop_thread(cfg.chain, stats::thread::name::LIVE_SYNC_BACKFILL_AND_PURGE);
     });
 
     // Live Fill data starts at fork 1.

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

@@ -4,7 +4,7 @@
 //! each network. Chain Followers use the data supplied by the Chain-Sync.
 //! This module configures the chain sync processes.
 
-use std::{panic, sync::LazyLock};
+use std::sync::LazyLock;
 
 use cardano_blockchain_types::Network;
 use dashmap::DashMap;
@@ -150,23 +150,13 @@ impl ChainSyncConfig {
         // Start the Mithril Snapshot Follower
         let rx = self.mithril_cfg.run().await?;
 
-        // Wrap inside a panic catcher to detect if the task panics.
-        let result = panic::catch_unwind(|| {
-            stats::start_thread(self.chain, stats::thread::name::CHAIN_SYNC, true);
-            // Start Chain Sync
-            tokio::spawn(chain_sync(self.clone(), rx))
-        });
-
-        if let Ok(handle) = result {
-            *locked_handle = Some(handle);
-        } else {
-            // Chain sync panic, stop the thread and log.
-            error!(
-                chain = self.chain.to_string(),
-                "Chain Sync for {} : PANICKED", self.chain
-            );
-            stats::stop_thread(self.chain, stats::thread::name::CHAIN_SYNC);
-        }
+        let config = self.clone();
+        // Start Chain Sync
+        *locked_handle = Some(tokio::spawn(async move {
+            stats::start_thread(config.chain, stats::thread::name::CHAIN_SYNC, true);
+            chain_sync(config.clone(), rx).await;
+            stats::stop_thread(config.chain, stats::thread::name::CHAIN_SYNC);
+        }));
 
         // sync_map.insert(chain, handle);
         debug!("Chain Sync for {} : Started", self.chain);

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

@@ -1,7 +1,6 @@
 //! Configuration for the Mithril Snapshot used by the follower.
 
 use std::{
-    panic,
     path::{Path, PathBuf},
     str::FromStr,
     sync::LazyLock,
@@ -415,27 +414,16 @@ impl MithrilSnapshotConfig {
         let (tx, rx) = mpsc::channel::<MithrilUpdateMessage>(2);
 
         // let handle = tokio::spawn(background_mithril_update(chain, self.clone(), tx));
-
-        // Wrap inside a panic catcher to detect if the task panics.
-        let result = panic::catch_unwind(|| {
+        let config = self.clone();
+        *locked_handle = Some(tokio::spawn(async move {
             stats::start_thread(
-                self.chain,
+                config.chain,
                 stats::thread::name::MITHRIL_SNAPSHOT_UPDATER,
                 true,
             );
-            tokio::spawn(background_mithril_update(self.clone(), tx))
-        });
-
-        if let Ok(handle) = result {
-            *locked_handle = Some(handle);
-        } else {
-            // Mithril update panic, stop the thread and log.
-            error!(
-                chain = self.chain.to_string(),
-                "Background Mithril Update for {} : PANICKED", self.chain
-            );
-            stats::stop_thread(self.chain, stats::thread::name::MITHRIL_SNAPSHOT_UPDATER);
-        }
+            background_mithril_update(config.clone(), tx).await;
+            stats::stop_thread(config.chain, stats::thread::name::MITHRIL_SNAPSHOT_UPDATER);
+        }));
 
         // sync_map.insert(chain, handle);
         debug!(

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

@@ -15,7 +15,6 @@ 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.
@@ -197,17 +196,7 @@ impl MithrilSnapshotIterator {
         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();
-            stats::start_thread(
-                inner_iterator.chain,
-                stats::thread::name::MITHRIL_ITERATOR_NEXT,
-                false,
-            );
-            let next = inner_iterator.next();
-            stats::stop_thread(
-                inner_iterator.chain,
-                stats::thread::name::MITHRIL_ITERATOR_NEXT,
-            );
-            next
+            inner_iterator.next()
         })
         .await;
 

rust/cardano-chain-follower/src/stats/thread/name.rs

@@ -10,8 +10,6 @@ pub(crate) const LIVE_SYNC_BACKFILL_AND_PURGE: &str = "LiveSyncBackfillAndPurge"
 pub(crate) const MITHRIL_ITERATOR: &str = "MithrilIterator";
 /// Background Mithril Snapshot Updater.
 pub(crate) const MITHRIL_SNAPSHOT_UPDATER: &str = "MithrilSnapshotUpdater";
-/// Mithril Snapshot Iterator Next.
-pub(crate) const MITHRIL_ITERATOR_NEXT: &str = "MithrilIteratorNext";
 /// Mithril compute snapshot.
 pub(crate) const COMPUTE_SNAPSHOT_MSG: &str = "ComputeSnapshotMsg";
 /// Background Mithril Snapshot Validator.

rust/cardano-chain-follower/src/turbo_downloader/mod.rs

@@ -421,7 +421,11 @@ impl ParallelDownloadProcessor {
             let (work_queue_tx, work_queue_rx) = crossbeam_channel::unbounded::<DlWorkOrder>();
             let params = self.0.clone();
             thread::spawn(move || {
+                let thread_name = format!("{}::{worker}", stats::thread::name::PARALLEL_DL_WORKER);
+
+                stats::start_thread(chain, thread_name.as_str(), false);
                 Self::worker(&params, worker, &work_queue_rx, chain);
+                stats::stop_thread(chain, thread_name.as_str());
             });
 
             let _unused = self.0.work_queue.insert(worker, work_queue_tx);
@@ -433,11 +437,13 @@ impl ParallelDownloadProcessor {
     /// Call the work queue receiver.
     /// This is a helper function to pause and resume the stats thread.
     fn call_work_queue_receiver(
-        chain: Network, name: &str, work_queue: &Receiver<usize>,
+        chain: Network, worker_id: usize, work_queue: &Receiver<usize>,
     ) -> Result<usize, RecvError> {
-        stats::pause_thread(chain, name);
+        let thread_name = format!("{}::{worker_id}", stats::thread::name::PARALLEL_DL_WORKER);
+
+        stats::pause_thread(chain, &thread_name);
         let recv = work_queue.recv();
-        stats::resume_thread(chain, name);
+        stats::resume_thread(chain, &thread_name);
         recv
     }
 
@@ -447,10 +453,6 @@ impl ParallelDownloadProcessor {
         params: &Arc<ParallelDownloadProcessorInner>, worker_id: usize,
         work_queue: &crossbeam_channel::Receiver<DlWorkOrder>, chain: Network,
     ) {
-        let thread_name = format!("{}::{worker_id}", stats::thread::name::PARALLEL_DL_WORKER);
-
-        stats::start_thread(chain, thread_name.as_str(), false);
-
         debug!("Worker {worker_id} started");
 
         // Each worker has its own http_client, so there is no cross worker pathology
@@ -463,9 +465,7 @@ impl ParallelDownloadProcessor {
         }
         let http_agent = params.cfg.make_http_agent(worker_id);
 
-        while let Ok(next_chunk) =
-            Self::call_work_queue_receiver(chain, thread_name.as_str(), work_queue)
-        {
+        while let Ok(next_chunk) = Self::call_work_queue_receiver(chain, worker_id, work_queue) {
             // Add a small delay to the first chunks for each worker.
             // So that the leading chunks are more likely to finish downloading first.
             if next_chunk > 0 && next_chunk < params.cfg.workers {
@@ -516,7 +516,6 @@ impl ParallelDownloadProcessor {
             // debug!("Worker {worker_id} DL chunk queued {next_chunk}");
         }
         debug!("Worker {worker_id} ended");
-        stats::stop_thread(chain, thread_name.as_str());
     }
 
     /// Send a work order to a worker.