feat: replace wrap_with_spawn with wrap_with_spawn_named to provide t…

…ask name
fleek-network · May 27, 2024 · 6246382 · 6246382
1 parent 555b88e
commit 6246382
Show file tree

Hide file tree

Showing 12 changed files with 61 additions and 30 deletions.
diff --git a/core/archive/src/archive.rs b/core/archive/src/archive.rs
@@ -39,9 +39,10 @@ struct ArchiveInner<C: Collection> {
 
 impl<C: Collection> BuildGraph for Archive<C> {
     fn build_graph() -> fdi::DependencyGraph {
-        fdi::DependencyGraph::new().with_infallible(
-            Self::new.with_event_handler("start", insertion_task::<C>.wrap_with_spawn()),
-        )
+        fdi::DependencyGraph::new().with_infallible(Self::new.with_event_handler(
+            "start",
+            insertion_task::<C>.wrap_with_spawn_named("ARCHIVE"),
+        ))
     }
 }
 

diff --git a/core/blockstore-server/src/blockstore_server.rs b/core/blockstore-server/src/blockstore_server.rs
@@ -85,8 +85,10 @@ impl<C: Collection> BlockstoreServer<C> {
 
 impl<C: Collection> BuildGraph for BlockstoreServer<C> {
     fn build_graph() -> fdi::DependencyGraph {
-        fdi::DependencyGraph::default()
-            .with(Self::init.with_event_handler("start", Self::start.wrap_with_spawn()))
+        fdi::DependencyGraph::default().with(Self::init.with_event_handler(
+            "start",
+            Self::start.wrap_with_spawn_named("BLOCKSTORE-SERVER"),
+        ))
     }
 }
 
@@ -165,7 +167,7 @@ impl<C: Collection> BlockstoreServerInner<C> {
                                                     Some("Counter for number of blockstore requests handled by this node")
                                                 );
                                             },
-                                            "BLOCKSTORE-SERVER: HANDLE-REQUEST"
+                                            "BLOCKSTORE-SERVER: handle request"
                                         );
                                     } else {
                                         self.num_responses.fetch_sub(1, Ordering::Release);

diff --git a/core/dack-aggregator/src/dack_aggregator.rs b/core/dack-aggregator/src/dack_aggregator.rs
@@ -53,8 +53,10 @@ impl<C: Collection> DeliveryAcknowledgmentAggregator<C> {
 
 impl<C: Collection> BuildGraph for DeliveryAcknowledgmentAggregator<C> {
     fn build_graph() -> fdi::DependencyGraph {
-        fdi::DependencyGraph::default()
-            .with(Self::init.with_event_handler("start", Self::start.wrap_with_spawn()))
+        fdi::DependencyGraph::default().with(Self::init.with_event_handler(
+            "start",
+            Self::start.wrap_with_spawn_named("DACK_AGGREGATOR"),
+        ))
     }
 }
 
@@ -151,14 +153,15 @@ impl AggregatorInner {
                             metadata: metadata.remove(&service_id),
                         };
                         let submit_tx = self.submit_tx.clone();
-                        tokio::spawn(async move {
+                        spawn!(async move {
                             if let Err(e) = submit_tx
                                 .run(update)
                                 .await
                             {
                                 error!("Failed to submit DACK to signer: {e:?}");
                             }
-                        });
+                        },
+                        "DACK_AGGREGATOR: submit tx");
                     }
 
                     increment_counter_by!(

diff --git a/core/handshake/src/handshake.rs b/core/handshake/src/handshake.rs
@@ -93,13 +93,19 @@ impl<C: Collection> Handshake<C> {
             if let Some(https) = this.config.https.clone() {
                 let https_router = router.clone();
                 let handle = run.handle.clone();
-                tokio::spawn(async move { spawn_https_server(https_router, https, handle).await });
+                spawn!(
+                    async move { spawn_https_server(https_router, https, handle).await },
+                    "HANDSHAKE: start optional http server"
+                );
             }
 
             // Start HTTP server.
             let waiter2 = waiter.clone();
             let http_addr = this.config.http_address;
-            tokio::spawn(async move { spawn_http_server(http_addr, router, waiter2).await });
+            spawn!(
+                async move { spawn_http_server(http_addr, router, waiter2).await },
+                "HANDSHAKE: start http server"
+            );
 
             // Shutdown the handle.
             waiter.wait_for_shutdown().await;
@@ -110,8 +116,9 @@ impl<C: Collection> Handshake<C> {
 
 impl<C: Collection> BuildGraph for Handshake<C> {
     fn build_graph() -> fdi::DependencyGraph {
-        fdi::DependencyGraph::new()
-            .with_infallible(Self::new.with_event_handler("start", Self::start.wrap_with_spawn()))
+        fdi::DependencyGraph::new().with_infallible(
+            Self::new.with_event_handler("start", Self::start.wrap_with_spawn_named("HANDSHAKE")),
+        )
     }
 }
 

diff --git a/core/interfaces/src/macros.rs b/core/interfaces/src/macros.rs
@@ -139,11 +139,11 @@ macro_rules! partial {
 #[macro_export]
 macro_rules! spawn {
     ($future:expr, $name:expr, crucial($waiter:expr)) => {
-        tokio::task::Builder::new().name(concat!($name,"#WAITER")).spawn(async move{
+        tokio::task::Builder::new().name(&format!("{}#WAITER", $name)).spawn(async move{
             let handle = tokio::task::Builder::new().name($name).spawn($future).expect("Tokio task created outside of tokio runtime");
 
             if let Err(e) = handle.await {
-                tracing::error!("Crucial task:{} had a panic: {:?} \n Signaling to shutdown the rest of the node",$name, e);
+                tracing::error!("Crucial task:{} had a panic: {:?} \n Signaling to shutdown the rest of the node", $name, e);
                 $crate::ShutdownWaiter::trigger_shutdown(&$waiter);
             }
         }).expect("Tokio task created outside of tokio runtime")

diff --git a/core/node/src/lib.rs b/core/node/src/lib.rs
@@ -82,15 +82,18 @@ impl<C: Collection> ContainedNode<C> {
     pub fn spawn(&self) -> JoinHandle<Result<()>> {
         let provider = self.provider.clone();
 
+        let waiter = self.shutdown.waiter();
         self.runtime.as_ref().unwrap().spawn_blocking(move || {
             let graph = C::build_graph();
             let mut provider = provider.get_local_provider();
 
             // Set tokio as the spawner of fdi async works.
             provider
                 .get_mut::<fdi::Executor>()
-                .set_spawn_cb(|fut, _name| {
-                    tokio::spawn(fut);
+                .set_spawn_cb(move |fut, name| {
+                    let name = name.expect("Name must be provided");
+                    let waiter = waiter.clone();
+                    spawn!(fut, &name, crucial(waiter));
                 });
 
             // Init all of the components and dependencies.

diff --git a/core/pinger/src/pinger.rs b/core/pinger/src/pinger.rs
@@ -64,8 +64,9 @@ impl<C: Collection> PingerInterface<C> for Pinger<C> {}
 
 impl<C: Collection> BuildGraph for Pinger<C> {
     fn build_graph() -> fdi::DependencyGraph {
-        fdi::DependencyGraph::new()
-            .with(Self::new.with_event_handler("start", Self::start.wrap_with_spawn()))
+        fdi::DependencyGraph::new().with(
+            Self::new.with_event_handler("start", Self::start.wrap_with_spawn_named("PINGER")),
+        )
     }
 }
 
@@ -203,12 +204,13 @@ impl<C: Collection> PingerInner<C> {
                         } else {
                             pending_req.insert((peer_index, id), Instant::now());
                             let tx = timeout_tx.clone();
-                            tokio::spawn(async move {
+                            spawn!(async move {
                                 tokio::time::sleep(TIMEOUT).await;
                                 // We ignore the sending error because it can happen that the
                                 // pinger is shutdown while there are still pending timeout tasks.
                                 let _ = tx.send((peer_index, id)).await;
-                            });
+                            },
+                            "PINGER: request timeout");
                         }
                     }
                 }

diff --git a/core/pool/src/provider.rs b/core/pool/src/provider.rs
@@ -150,7 +150,7 @@ where
 impl<C: Collection> BuildGraph for PoolProvider<C, QuinnMuxer> {
     fn build_graph() -> fdi::DependencyGraph {
         fdi::DependencyGraph::new()
-            .with(Self::init.with_event_handler("start", Self::start.wrap_with_spawn()))
+            .with(Self::init.with_event_handler("start", Self::start.wrap_with_spawn_named("POOL")))
     }
 }
 

diff --git a/core/rep-collector/src/aggregator.rs b/core/rep-collector/src/aggregator.rs
@@ -211,8 +211,14 @@ impl<C: Collection> ReputationAggregator<C> {
 
 impl<C: Collection> BuildGraph for ReputationAggregator<C> {
     fn build_graph() -> fdi::DependencyGraph {
-        fdi::DependencyGraph::new()
-            .with(Self::new.with_event_handler("start", Self::start.bounded().wrap_with_spawn()))
+        fdi::DependencyGraph::new().with(
+            Self::new.with_event_handler(
+                "start",
+                Self::start
+                    .bounded()
+                    .wrap_with_spawn_named("REP-AGGREGATOR"),
+            ),
+        )
     }
 }
 

diff --git a/core/resolver/src/resolver.rs b/core/resolver/src/resolver.rs
@@ -70,8 +70,9 @@ impl<C: Collection> Resolver<C> {
 
 impl<C: Collection> BuildGraph for Resolver<C> {
     fn build_graph() -> fdi::DependencyGraph {
-        fdi::DependencyGraph::default()
-            .with(Self::init.with_event_handler("start", Self::start.wrap_with_spawn()))
+        fdi::DependencyGraph::default().with(
+            Self::init.with_event_handler("start", Self::start.wrap_with_spawn_named("RESOLVER")),
+        )
     }
 }
 

diff --git a/core/test-utils/src/consensus.rs b/core/test-utils/src/consensus.rs
@@ -152,7 +152,12 @@ impl<C: Collection> BuildGraph for MockConsensus<C> {
             .with_infallible(|config: fdi::Ref<C::ConfigProviderInterface>| {
                 MockConsensusGroup::new(config.get::<Self>())
             })
-            .with_infallible(Self::new.with_event_handler("start", Self::start.wrap_with_spawn()))
+            .with_infallible(
+                Self::new.with_event_handler(
+                    "start",
+                    Self::start.wrap_with_spawn_named("MOCK-CONSENSUS"),
+                ),
+            )
     }
 }
 

diff --git a/core/topology/src/lib.rs b/core/topology/src/lib.rs
@@ -130,8 +130,9 @@ impl<C: Collection> Topology<C> {
 
 impl<C: Collection> BuildGraph for Topology<C> {
     fn build_graph() -> fdi::DependencyGraph {
-        fdi::DependencyGraph::default()
-            .with(Self::init.with_event_handler("start", Self::start.wrap_with_spawn()))
+        fdi::DependencyGraph::default().with(
+            Self::init.with_event_handler("start", Self::start.wrap_with_spawn_named("TOPOLOGY")),
+        )
     }
 }