--wip--

Author: rmburg

Cargo.toml

@@ -133,6 +133,9 @@ edition = "2024"
 license = "GPL-3.0-only"
 homepage = "https://github.com/hulks/hulk"
 
+[workspace.lints.rust]
+unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)'] }
+
 [workspace.dependencies]
 active_vision = { path = "crates/nodes/active_vision" }
 aliveness = { path = "crates/aliveness" }

crates/hulk_ros_z/Cargo.toml

@@ -5,6 +5,9 @@ edition.workspace = true
 license.workspace = true
 homepage.workspace = true
 
+[lints]
+workspace = true
+
 [package.metadata.pepsi]
 cross-compile = true
 

crates/hulk_ros_z/src/main.rs

@@ -214,7 +214,28 @@ where
     F: Future<Output = Result<()>> + Send + 'static,
 {
     let span = tracing::info_span!(target: "runtime::hulk_ros_z", "hulk_ros_z_node", node);
-    join_set.spawn(future.instrument(span));
+    spawn_join_set_task(join_set, node, future.instrument(span));
+}
+
+#[cfg(tokio_unstable)]
+fn spawn_join_set_task<F>(join_set: &mut JoinSet<Result<()>>, node: &'static str, future: F)
+where
+    F: Future<Output = Result<()>> + Send + 'static,
+{
+    join_set
+        .build_task()
+        .name(node)
+        .spawn(future)
+        .unwrap_or_else(|error| panic!("failed to spawn {node} task: {error}"));
+}
+
+#[cfg(not(tokio_unstable))]
+fn spawn_join_set_task<F>(join_set: &mut JoinSet<Result<()>>, node: &'static str, future: F)
+where
+    F: Future<Output = Result<()>> + Send + 'static,
+{
+    let _ = node;
+    join_set.spawn(future);
 }
 
 async fn monitor(join_set: &mut JoinSet<Result<()>>) -> Result<()> {

crates/ros-z/Cargo.toml

@@ -8,6 +8,9 @@ homepage.workspace = true
 [lib]
 crate-type = ["lib"]
 
+[lints]
+workspace = true
+
 [dependencies]
 arc-swap = { workspace = true }
 byteorder = { workspace = true }

crates/ros-z/src/cache.rs

@@ -37,6 +37,7 @@
 
 use parking_lot::RwLock;
 use std::collections::{BTreeMap, VecDeque};
+use std::future::Future;
 use std::marker::PhantomData;
 use std::sync::Arc;
 use tracing::{Instrument, debug, warn};
@@ -480,6 +481,7 @@ where
         let inner_cb = inner.clone();
         let node = sub_builder.context.node.fully_qualified_name();
         let topic = sub_builder.topic.clone();
+        let task_name = cache_task_name(&node, &topic, "zenoh");
 
         let raw_subscriber = sub_builder.raw().build().await?;
         let mut raw_subscriber_task = raw_subscriber;
@@ -490,7 +492,8 @@ where
             topic = %topic,
             stamp = "zenoh"
         );
-        let task = tokio::spawn(
+        let task = spawn_cache_task(
+            &task_name,
             async move {
                 loop {
                     let sample = match raw_subscriber_task.recv().await {
@@ -570,6 +573,7 @@ where
         let inner_cb = inner.clone();
         let node = sub_builder.context.node.fully_qualified_name();
         let topic = sub_builder.topic.clone();
+        let task_name = cache_task_name(&node, &topic, "extractor");
 
         let raw_subscriber = sub_builder.raw().build().await?;
         let mut raw_subscriber_task = raw_subscriber;
@@ -580,7 +584,8 @@ where
             topic = %topic,
             stamp = "extractor"
         );
-        let task = tokio::spawn(
+        let task = spawn_cache_task(
+            &task_name,
             async move {
                 loop {
                     let sample = match raw_subscriber_task.recv().await {
@@ -611,6 +616,32 @@ where
     }
 }
 
+fn cache_task_name(node: &str, topic: &str, stamp: &str) -> String {
+    format!("ros_z_cache:{node}:{topic}:{stamp}")
+}
+
+#[cfg(tokio_unstable)]
+fn spawn_cache_task<F>(name: &str, future: F) -> tokio::task::JoinHandle<F::Output>
+where
+    F: Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    tokio::task::Builder::new()
+        .name(name)
+        .spawn(future)
+        .unwrap_or_else(|error| panic!("failed to spawn {name} task: {error}"))
+}
+
+#[cfg(not(tokio_unstable))]
+fn spawn_cache_task<F>(name: &str, future: F) -> tokio::task::JoinHandle<F::Output>
+where
+    F: Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    let _ = name;
+    tokio::spawn(future)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;