watched_tasks: spawn long running tasks via special task builder

The WatchedTasksBuilder makes sure that the program ends if any of the
tasks fails and also handles error propagation from the tasks to
the main() function.

Signed-off-by: Leonard Göhrs <l.goehrs@pengutronix.de>

Author: hnez

src/adc.rs

@@ -19,10 +19,11 @@ use std::time::Duration;
 
 use anyhow::Result;
 use async_std::sync::Arc;
-use async_std::task::{sleep, spawn};
+use async_std::task::sleep;
 
 use crate::broker::{BrokerBuilder, Topic};
 use crate::measurement::{Measurement, Timestamp};
+use crate::watched_tasks::WatchedTasksBuilder;
 
 const HISTORY_LENGTH: usize = 200;
 const SLOW_INTERVAL: Duration = Duration::from_millis(100);
@@ -77,7 +78,7 @@ pub struct Adc {
 }
 
 impl Adc {
-    pub async fn new(bb: &mut BrokerBuilder) -> Result<Self> {
+    pub async fn new(bb: &mut BrokerBuilder, wtb: &mut WatchedTasksBuilder) -> Result<Self> {
         let stm32_thread = IioThread::new_stm32().await?;
         let powerboard_thread = IioThread::new_powerboard().await?;
 
@@ -212,7 +213,7 @@ impl Adc {
 
         // Spawn an async task to transfer values from the Atomic value based
         // "fast" interface to the broker based "slow" interface.
-        spawn(async move {
+        wtb.spawn_task("adc-update", async move {
             loop {
                 sleep(SLOW_INTERVAL).await;
 

src/backlight.rs

@@ -18,7 +18,6 @@
 use anyhow::Result;
 use async_std::prelude::*;
 use async_std::sync::Arc;
-use async_std::task::spawn;
 use log::warn;
 
 mod demo_mode;
@@ -30,21 +29,22 @@ use demo_mode::{Backlight as SysBacklight, Brightness, SysClass};
 use sysfs_class::{Backlight as SysBacklight, Brightness, SysClass};
 
 use crate::broker::{BrokerBuilder, Topic};
+use crate::watched_tasks::WatchedTasksBuilder;
 
 pub struct Backlight {
     pub brightness: Arc<Topic<f32>>,
 }
 
 impl Backlight {
-    pub fn new(bb: &mut BrokerBuilder) -> Result<Self> {
+    pub fn new(bb: &mut BrokerBuilder, wtb: &mut WatchedTasksBuilder) -> Result<Self> {
         let brightness = bb.topic_rw("/v1/tac/display/backlight/brightness", Some(1.0));
 
         let (mut rx, _) = brightness.clone().subscribe_unbounded();
 
         let backlight = SysBacklight::new("backlight")?;
         let max_brightness = backlight.max_brightness()?;
 
-        spawn(async move {
+        wtb.spawn_task("backlight-dimmer", async move {
             while let Some(fraction) = rx.next().await {
                 let brightness = (max_brightness as f32) * fraction;
                 let mut brightness = brightness.clamp(0.0, max_brightness as f32) as u64;
@@ -60,6 +60,8 @@ impl Backlight {
                     warn!("Failed to set LED pattern: {}", e);
                 }
             }
+
+            Ok(())
         });
 
         Ok(Self { brightness })

src/broker.rs

@@ -18,6 +18,8 @@
 use async_std::sync::Arc;
 use serde::{de::DeserializeOwned, Serialize};
 
+use crate::watched_tasks::WatchedTasksBuilder;
+
 mod mqtt_conn;
 mod persistence;
 mod rest;
@@ -113,10 +115,10 @@ impl BrokerBuilder {
     /// Finish building the broker
     ///
     /// This consumes the builder so that no new topics can be registered
-    pub fn build(self, server: &mut tide::Server<()>) {
+    pub fn build(self, wtb: &mut WatchedTasksBuilder, server: &mut tide::Server<()>) {
         let topics = Arc::new(self.topics);
 
-        persistence::register(topics.clone());
+        persistence::register(wtb, topics.clone());
         rest::register(server, topics.clone());
         mqtt_conn::register(server, topics);
     }

src/broker/persistence.rs

@@ -22,13 +22,14 @@ use anyhow::{bail, Result};
 use async_std::channel::{unbounded, Receiver};
 use async_std::prelude::*;
 use async_std::sync::Arc;
-use async_std::task::spawn;
 use log::{error, info};
 use serde::{Deserialize, Serialize};
 use serde_json::{from_reader, to_writer_pretty, Map, Value};
 
 use super::{AnyTopic, TopicName};
 
+use crate::watched_tasks::WatchedTasksBuilder;
+
 #[cfg(feature = "demo_mode")]
 const PERSISTENCE_PATH: &str = "demo_files/srv/tacd/state.json";
 
@@ -147,7 +148,7 @@ async fn save_on_change(
     Ok(())
 }
 
-pub fn register(topics: Arc<Vec<Arc<dyn AnyTopic>>>) {
+pub fn register(wtb: &mut WatchedTasksBuilder, topics: Arc<Vec<Arc<dyn AnyTopic>>>) {
     load(&topics).unwrap();
 
     let (tx, rx) = unbounded();
@@ -156,5 +157,5 @@ pub fn register(topics: Arc<Vec<Arc<dyn AnyTopic>>>) {
         topic.subscribe_as_bytes(tx.clone(), false);
     }
 
-    spawn(async move { save_on_change(topics, rx).await.unwrap() });
+    wtb.spawn_task("persistence-save", save_on_change(topics, rx));
 }

src/dbus.rs

@@ -19,6 +19,7 @@ use async_std::sync::Arc;
 
 use crate::broker::{BrokerBuilder, Topic};
 use crate::led::BlinkPattern;
+use crate::watched_tasks::WatchedTasksBuilder;
 
 #[cfg(feature = "demo_mode")]
 mod zb {
@@ -78,6 +79,7 @@ pub struct DbusSession {
 impl DbusSession {
     pub async fn new(
         bb: &mut BrokerBuilder,
+        wtb: &mut WatchedTasksBuilder,
         led_dut: Arc<Topic<BlinkPattern>>,
         led_uplink: Arc<Topic<BlinkPattern>>,
     ) -> Self {
@@ -91,10 +93,10 @@ impl DbusSession {
         let conn = Arc::new(tacd.serve(conn_builder).build().await.unwrap());
 
         Self {
-            hostname: Hostname::new(bb, &conn),
-            network: Network::new(bb, &conn, led_dut, led_uplink),
-            rauc: Rauc::new(bb, &conn),
-            systemd: Systemd::new(bb, &conn).await,
+            hostname: Hostname::new(bb, wtb, &conn),
+            network: Network::new(bb, wtb, &conn, led_dut, led_uplink),
+            rauc: Rauc::new(bb, wtb, &conn),
+            systemd: Systemd::new(bb, wtb, &conn).await,
         }
     }
 }

src/dbus/hostname.rs

@@ -21,9 +21,11 @@ use async_std::stream::StreamExt;
 #[cfg(not(feature = "demo_mode"))]
 use zbus::Connection;
 
-use crate::broker::{BrokerBuilder, Topic};
 use async_std::sync::Arc;
 
+use crate::broker::{BrokerBuilder, Topic};
+use crate::watched_tasks::WatchedTasksBuilder;
+
 mod hostnamed;
 
 pub struct Hostname {
@@ -32,19 +34,24 @@ pub struct Hostname {
 
 impl Hostname {
     #[cfg(feature = "demo_mode")]
-    pub fn new<C>(bb: &mut BrokerBuilder, _conn: C) -> Self {
+    pub fn new<C>(bb: &mut BrokerBuilder, _wtb: &mut WatchedTasksBuilder, _conn: C) -> Self {
         Self {
             hostname: bb.topic_ro("/v1/tac/network/hostname", Some("lxatac".into())),
         }
     }
 
     #[cfg(not(feature = "demo_mode"))]
-    pub fn new(bb: &mut BrokerBuilder, conn: &Arc<Connection>) -> Self {
+    pub fn new(
+        bb: &mut BrokerBuilder,
+        wtb: &mut WatchedTasksBuilder,
+        conn: &Arc<Connection>,
+    ) -> Self {
         let hostname = bb.topic_ro("/v1/tac/network/hostname", None);
 
         let conn = conn.clone();
         let hostname_topic = hostname.clone();
-        async_std::task::spawn(async move {
+
+        wtb.spawn_task("hostname-update", async move {
             let proxy = hostnamed::HostnameProxy::new(&conn).await.unwrap();
 
             let mut stream = proxy.receive_hostname_changed().await;
@@ -58,6 +65,8 @@ impl Hostname {
                     hostname_topic.set(h);
                 }
             }
+
+            Ok(())
         });
 
         Self { hostname }

src/dbus/networkmanager.rs

@@ -22,6 +22,7 @@ use serde::{Deserialize, Serialize};
 
 use crate::broker::{BrokerBuilder, Topic};
 use crate::led::BlinkPattern;
+use crate::watched_tasks::WatchedTasksBuilder;
 
 // Macro use makes these modules quite heavy, so we keep them commented
 // out until they are actually used
@@ -244,6 +245,7 @@ impl Network {
     #[cfg(feature = "demo_mode")]
     pub fn new<C>(
         bb: &mut BrokerBuilder,
+        _wtb: &mut WatchedTasksBuilder,
         _conn: C,
         _led_dut: Arc<Topic<BlinkPattern>>,
         _led_uplink: Arc<Topic<BlinkPattern>>,
@@ -266,6 +268,7 @@ impl Network {
     #[cfg(not(feature = "demo_mode"))]
     pub fn new(
         bb: &mut BrokerBuilder,
+        wtb: &mut WatchedTasksBuilder,
         conn: &Arc<Connection>,
         led_dut: Arc<Topic<BlinkPattern>>,
         led_uplink: Arc<Topic<BlinkPattern>>,
@@ -274,26 +277,20 @@ impl Network {
 
         let conn_task = conn.clone();
         let dut_interface = this.dut_interface.clone();
-        async_std::task::spawn(async move {
-            handle_link_updates(&conn_task, dut_interface, "dut", led_dut)
-                .await
-                .unwrap();
+        wtb.spawn_task("link-dut-update", async move {
+            handle_link_updates(&conn_task, dut_interface, "dut", led_dut).await
         });
 
         let conn_task = conn.clone();
         let uplink_interface = this.uplink_interface.clone();
-        async_std::task::spawn(async move {
-            handle_link_updates(&conn_task, uplink_interface, "uplink", led_uplink)
-                .await
-                .unwrap();
+        wtb.spawn_task("link-uplink-update", async move {
+            handle_link_updates(&conn_task, uplink_interface, "uplink", led_uplink).await
         });
 
         let conn_task = conn.clone();
         let bridge_interface = this.bridge_interface.clone();
-        async_std::task::spawn(async move {
-            handle_ipv4_updates(&conn_task, bridge_interface, "tac-bridge")
-                .await
-                .unwrap();
+        wtb.spawn_task("ip-tac-bridge-update", async move {
+            handle_ipv4_updates(&conn_task, bridge_interface, "tac-bridge").await
         });
 
         this