watched_tasks: also handle threads

Execution should stop if any of the important threads ends, not just
one of the tasks.

Signed-off-by: Leonard Göhrs <[email protected]>

Author: hnez

src/adc.rs

@@ -79,8 +79,8 @@ pub struct Adc {
 
 impl Adc {
     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?;
+        let stm32_thread = IioThread::new_stm32(wtb).await?;
+        let powerboard_thread = IioThread::new_powerboard(wtb).await?;
 
         let adc = Self {
             usb_host_curr: AdcChannel {

src/adc/iio/demo_mode.rs

@@ -160,7 +160,7 @@ pub struct IioThread {
 }
 
 impl IioThread {
-    pub async fn new_stm32() -> Result<Arc<Self>> {
+    pub async fn new_stm32<W>(_wtb: &W) -> Result<Arc<Self>> {
         let mut demo_magic = block_on(DEMO_MAGIC_STM32.lock());
 
         // Only ever set up a single demo_mode "IioThread" per ADC
@@ -195,7 +195,7 @@ impl IioThread {
         Ok(this)
     }
 
-    pub async fn new_powerboard() -> Result<Arc<Self>> {
+    pub async fn new_powerboard<W>(_wtb: &W) -> Result<Arc<Self>> {
         let mut demo_magic = block_on(DEMO_MAGIC_POWERBOARD.lock());
 
         // Only ever set up a single demo_mode "IioThread" per ADC

src/adc/iio/hardware.rs

@@ -20,9 +20,6 @@ use std::fs::create_dir;
 use std::io::Read;
 use std::path::Path;
 use std::sync::atomic::{AtomicU16, AtomicU64, Ordering};
-use std::sync::Mutex;
-use std::thread;
-use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 
 use anyhow::{anyhow, Context, Error, Result};
@@ -35,6 +32,7 @@ use log::{debug, error, warn};
 use thread_priority::*;
 
 use crate::measurement::{Measurement, Timestamp};
+use crate::watched_tasks::WatchedTasksBuilder;
 
 struct ChannelDesc {
     kernel_name: &'static str,
@@ -255,7 +253,6 @@ pub struct IioThread {
     ref_instant: Instant,
     timestamp: AtomicU64,
     values: Vec<AtomicU16>,
-    join: Mutex<Option<JoinHandle<()>>>,
     channel_descs: &'static [ChannelDesc],
 }
 
@@ -325,7 +322,8 @@ impl IioThread {
     }
 
     async fn new(
-        thread_name: &str,
+        wtb: &mut WatchedTasksBuilder,
+        thread_name: &'static str,
         adc_name: &'static str,
         trigger_name: &'static str,
         sample_rate: i64,
@@ -342,9 +340,8 @@ impl IioThread {
         let (thread_res_tx, thread_res_rx) = bounded(1);
 
         // Spawn a high priority thread that updates the atomic values in `thread`.
-        let join = thread::Builder::new()
-            .name(format!("tacd {thread_name} iio"))
-            .spawn(move || {
+        wtb.spawn_thread(thread_name, move || {
+            {
                 let adc_setup_res = Self::adc_setup(
                     adc_name,
                     trigger_name,
@@ -358,17 +355,15 @@ impl IioThread {
                             ref_instant: Instant::now(),
                             timestamp: AtomicU64::new(TIMESTAMP_ERROR),
                             values: channels.iter().map(|_| AtomicU16::new(0)).collect(),
-                            join: Mutex::new(None),
                             channel_descs,
                         });
-
                         (thread, channels, buf)
                     }
                     Err(e) => {
                         // Can not fail in practice as the queue is known to be empty
                         // at this point.
                         thread_res_tx.try_send(Err(e)).unwrap();
-                        return;
+                        return Ok(());
                     }
                 };
 
@@ -423,21 +418,20 @@ impl IioThread {
                         tx.try_send(Ok(content)).unwrap();
                     }
                 }
-            })?;
+            };
 
-        let thread = thread_res_rx.recv().await??;
+            Ok(())
+        })?;
 
-        // Locking the Mutex could only fail if the Mutex was poisoned by
-        // a thread that held the lock and panicked.
-        // At this point the Mutex has not yet been locked in another thread.
-        *thread.join.lock().unwrap() = Some(join);
+        let thread = thread_res_rx.recv().await??;
 
         Ok(thread)
     }
 
-    pub async fn new_stm32() -> Result<Arc<Self>> {
+    pub async fn new_stm32(wtb: &mut WatchedTasksBuilder) -> Result<Arc<Self>> {
         Self::new(
-            "stm32",
+            wtb,
+            "adc-stm32",
             "48003000.adc:adc@0",
             "tim4_trgo",
             80,
@@ -447,14 +441,23 @@ impl IioThread {
         .await
     }
 
-    pub async fn new_powerboard() -> Result<Arc<Self>> {
+    pub async fn new_powerboard(wtb: &mut WatchedTasksBuilder) -> Result<Arc<Self>> {
         let hr_trigger_path = Path::new(TRIGGER_HR_PWR_DIR);
 
         if !hr_trigger_path.is_dir() {
             create_dir(hr_trigger_path)?;
         }
 
-        Self::new("powerboard", "lmp92064", "tacd-pwr", 20, CHANNELS_PWR, 1).await
+        Self::new(
+            wtb,
+            "adc-powerboard",
+            "lmp92064",
+            "tacd-pwr",
+            20,
+            CHANNELS_PWR,
+            1,
+        )
+        .await
     }
 
     /// Use the channel names defined at the top of the file to get a reference

src/adc/iio/test.rs

@@ -107,7 +107,7 @@ pub struct IioThread {
 }
 
 impl IioThread {
-    pub async fn new_stm32() -> Result<Arc<Self>> {
+    pub async fn new_stm32<W>(_wtb: &W) -> Result<Arc<Self>> {
         let mut channels = Vec::new();
 
         for name in CHANNELS_STM32 {
@@ -117,7 +117,7 @@ impl IioThread {
         Ok(Arc::new(Self { channels }))
     }
 
-    pub async fn new_powerboard() -> Result<Arc<Self>> {
+    pub async fn new_powerboard<W>(_wtb: &W) -> Result<Arc<Self>> {
         let mut channels = Vec::new();
 
         for name in CHANNELS_PWR {

src/digital_io/gpio/demo_mode.rs

@@ -32,8 +32,8 @@ impl LineHandle {
         // It is just a hack to let adc/iio/demo_mode.rs
         // communicate with this function so that toggling an output
         // has an effect on the measured values.
-        let iio_thread_stm32 = block_on(IioThread::new_stm32()).unwrap();
-        let iio_thread_pwr = block_on(IioThread::new_powerboard()).unwrap();
+        let iio_thread_stm32 = block_on(IioThread::new_stm32(&())).unwrap();
+        let iio_thread_pwr = block_on(IioThread::new_powerboard(&())).unwrap();
 
         match self.name.as_str() {
             "OUT_0" => iio_thread_stm32

src/dut_power.rs

@@ -320,9 +320,8 @@ impl DutPwrThread {
 
         // Spawn a high priority thread that handles the power status
         // in a realtimey fashion.
-        thread::Builder::new()
-            .name("tacd power".into())
-            .spawn(move || {
+        wtb.spawn_thread("power-thread", move || {
+            {
                 let mut last_ts: Option<Instant> = None;
 
                 // There may be transients in the measured voltage/current, e.g. due to EMI or
@@ -482,7 +481,10 @@ impl DutPwrThread {
 
                 // Make sure to enter fail safe mode before leaving the thread
                 turn_off_with_reason(OutputState::Off, &pwr_line, &discharge_line, &state);
-            })?;
+            };
+
+            Ok(())
+        })?;
 
         let (tick, request, state) = thread_res_rx.next().await.unwrap()?;
 

src/main.rs

@@ -82,7 +82,7 @@ async fn init() -> Result<(Ui, HttpServer, WatchedTasksBuilder)> {
     .await?;
     let dig_io = DigitalIo::new(&mut bb, &mut wtb, led.out_0.clone(), led.out_1.clone());
     let regulators = Regulators::new(&mut bb, &mut wtb);
-    let temperatures = Temperatures::new(&mut bb);
+    let temperatures = Temperatures::new(&mut bb, &mut wtb)?;
     let usb_hub = UsbHub::new(
         &mut bb,
         &mut wtb,

src/regulators.rs

@@ -31,7 +31,7 @@ mod reg {
 
     pub fn regulator_set(name: &str, state: bool) -> Result<()> {
         if name == "output_iobus_12v" {
-            let iio_thread = block_on(IioThread::new_stm32()).unwrap();
+            let iio_thread = block_on(IioThread::new_stm32(&())).unwrap();
 
             iio_thread
                 .clone()

src/temperatures.rs

@@ -19,34 +19,37 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::thread::sleep;
 use std::time::Duration;
 
+use anyhow::Result;
 use async_std::sync::Arc;
-use async_std::task::spawn_blocking;
 use serde::{Deserialize, Serialize};
 
 use crate::broker::{BrokerBuilder, Topic};
 use crate::measurement::Measurement;
+use crate::watched_tasks::WatchedTasksBuilder;
 
 #[cfg(feature = "demo_mode")]
 mod hw {
+    use anyhow::Result;
+
     pub(super) trait SysClass {
-        fn input(&self) -> Result<u32, ()>;
+        fn input(&self) -> Result<u32>;
     }
 
     pub(super) struct HwMon;
     pub(super) struct TempDecoy;
 
     impl SysClass for TempDecoy {
-        fn input(&self) -> Result<u32, ()> {
+        fn input(&self) -> Result<u32> {
             Ok(30_000)
         }
     }
 
     impl HwMon {
-        pub(super) fn new(_: &'static str) -> Result<Self, ()> {
+        pub(super) fn new(_: &'static str) -> Result<Self> {
             Ok(Self)
         }
 
-        pub(super) fn temp(&self, _: u64) -> Result<TempDecoy, ()> {
+        pub(super) fn temp(&self, _: u64) -> Result<TempDecoy> {
             Ok(TempDecoy)
         }
     }
@@ -89,7 +92,7 @@ pub struct Temperatures {
 }
 
 impl Temperatures {
-    pub fn new(bb: &mut BrokerBuilder) -> Self {
+    pub fn new(bb: &mut BrokerBuilder, wtb: &mut WatchedTasksBuilder) -> Result<Self> {
         let run = Arc::new(AtomicBool::new(true));
         let soc_temperature = bb.topic_ro("/v1/tac/temperatures/soc", None);
         let warning = bb.topic_ro("/v1/tac/temperatures/warning", None);
@@ -98,14 +101,9 @@ impl Temperatures {
         let soc_temperature_thread = soc_temperature.clone();
         let warning_thread = warning.clone();
 
-        spawn_blocking(move || {
+        wtb.spawn_thread("temperature-update", move || {
             while run_thread.load(Ordering::Relaxed) {
-                let val = HwMon::new("hwmon0")
-                    .unwrap()
-                    .temp(1)
-                    .unwrap()
-                    .input()
-                    .unwrap();
+                let val = HwMon::new("hwmon0")?.temp(1)?.input()?;
 
                 let val = val as f32 / 1000.0;
 
@@ -120,13 +118,15 @@ impl Temperatures {
 
                 sleep(UPDATE_INTERVAL);
             }
-        });
 
-        Self {
+            Ok(())
+        })?;
+
+        Ok(Self {
             soc_temperature,
             warning,
             run: Some(run),
-        }
+        })
     }
 }
 

src/ui.rs

@@ -141,9 +141,10 @@ impl Ui {
         let screens = screens::init(wtb, &res, &alerts, &buttons, &reboot_message, &locator);
 
         handle_buttons(
+            wtb,
             "/dev/input/by-path/platform-gpio-keys-event",
             buttons.clone(),
-        );
+        )?;
 
         // Blink the status LED when locator is active
         let led_status_pattern = res.led.status.clone();

src/ui/buttons.rs

@@ -17,11 +17,13 @@
 
 use std::time::Duration;
 
+use anyhow::Result;
 use async_std::sync::Arc;
-use async_std::task::{block_on, sleep, spawn, spawn_blocking, JoinHandle};
+use async_std::task::{block_on, sleep, spawn, JoinHandle};
 use serde::{Deserialize, Serialize};
 
 use crate::broker::Topic;
+use crate::watched_tasks::WatchedTasksBuilder;
 
 pub const LONG_PRESS: Duration = Duration::from_millis(500);
 
@@ -133,8 +135,12 @@ impl ButtonEvent {
 
 /// Spawn a thread that blockingly reads user input and pushes them into
 /// a broker framework topic.
-pub fn handle_buttons(path: &'static str, topic: Arc<Topic<ButtonEvent>>) {
-    spawn_blocking(move || {
+pub fn handle_buttons(
+    wtb: &mut WatchedTasksBuilder,
+    path: &'static str,
+    topic: Arc<Topic<ButtonEvent>>,
+) -> Result<()> {
+    wtb.spawn_thread("button-input-thread", move || {
         let mut device = Device::open(path).unwrap();
         let mut press_task: [Option<JoinHandle<()>>; 2] = [None, None];
         let mut start_time = [None, None];
@@ -179,5 +185,7 @@ pub fn handle_buttons(path: &'static str, topic: Arc<Topic<ButtonEvent>>) {
                 }
             }
         }
-    });
+    })?;
+
+    Ok(())
 }

src/usb_hub.rs

@@ -94,7 +94,7 @@ mod rw {
 
         for (path_tail, iio_channel) in DISABLE_CHANNELS {
             if path.ends_with(path_tail) {
-                let iio_thread = block_on(IioThread::new_stm32()).unwrap();
+                let iio_thread = block_on(IioThread::new_stm32(&())).unwrap();
 
                 iio_thread
                     .get_channel(iio_channel)