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 {
@@ -74,6 +75,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 {
@@ -87,9 +89,9 @@ impl DbusSession {
         let conn = Arc::new(tacd.serve(conn_builder).build().await.unwrap());
 
         Self {
-            network: Network::new(bb, &conn, led_dut, led_uplink),
-            rauc: Rauc::new(bb, &conn),
-            systemd: Systemd::new(bb, &conn).await,
+            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/networkmanager/mod.rs

@@ -22,6 +22,7 @@ use serde::{Deserialize, Serialize};
 
 use crate::broker::{BrokerBuilder, Topic};
 use crate::led::BlinkPattern;
+use crate::watched_tasks::WatchedTasksBuilder;
 
 mod devices;
 mod hostname;
@@ -256,6 +257,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>>,
@@ -279,6 +281,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>>,
@@ -288,7 +291,7 @@ impl Network {
         {
             let conn = conn.clone();
             let hostname_topic = this.hostname.clone();
-            async_std::task::spawn(async move {
+            wtb.spawn_task("hostname-update", async move {
                 let proxy = hostname::HostnameProxy::new(&conn).await.unwrap();
 
                 let mut stream = proxy.receive_hostname_changed().await;
@@ -302,13 +305,15 @@ impl Network {
                         hostname_topic.set(h);
                     }
                 }
+
+                Ok(())
             });
         }
 
         {
             let conn = conn.clone();
             let dut_interface = this.dut_interface.clone();
-            async_std::task::spawn(async move {
+            wtb.spawn_task("link-dut-update", async move {
                 let mut link_stream = loop {
                     if let Ok(ls) = LinkStream::new(conn.clone(), "dut").await {
                         break ls;
@@ -330,13 +335,15 @@ impl Network {
 
                     dut_interface.set(info);
                 }
+
+                Ok(())
             });
         }
 
         {
             let conn = conn.clone();
             let uplink_interface = this.uplink_interface.clone();
-            async_std::task::spawn(async move {
+            wtb.spawn_task("link-uplink-update", async move {
                 let mut link_stream = loop {
                     if let Ok(ls) = LinkStream::new(conn.clone(), "uplink").await {
                         break ls;
@@ -355,13 +362,15 @@ impl Network {
 
                     uplink_interface.set(info);
                 }
+
+                Ok(())
             });
         }
 
         {
             let conn = conn.clone();
             let bridge_interface = this.bridge_interface.clone();
-            async_std::task::spawn(async move {
+            wtb.spawn_task("ip-tac-bridge-update", async move {
                 let mut ip_stream = loop {
                     if let Ok(ips) = IpStream::new(conn.clone(), "tac-bridge").await {
                         break ips;
@@ -375,6 +384,8 @@ impl Network {
                 while let Ok(info) = ip_stream.next(&conn).await {
                     bridge_interface.set(info);
                 }
+
+                Ok(())
             });
         }