adc: iio: make the iio thread less panicy

Previously the iio thread code contained a lot of .unwrap() calls and
explicit panics. Now it contains a few less and gained some comments
where it still .unwrap()s.

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

Author: hnez

src/adc.rs

@@ -195,50 +195,36 @@ impl Adc {
             time: bb.topic_ro("/v1/tac/time/now", None),
         };
 
-        let adc_clone = adc.clone();
+        let channels = [
+            adc.usb_host_curr.clone(),
+            adc.usb_host1_curr.clone(),
+            adc.usb_host2_curr.clone(),
+            adc.usb_host3_curr.clone(),
+            adc.out0_volt.clone(),
+            adc.out1_volt.clone(),
+            adc.iobus_curr.clone(),
+            adc.iobus_volt.clone(),
+            adc.pwr_volt.clone(),
+            adc.pwr_curr.clone(),
+        ];
+
+        let time = adc.time.clone();
 
         // Spawn an async task to transfer values from the Atomic value based
         // "fast" interface to the broker based "slow" interface.
         spawn(async move {
             loop {
                 sleep(SLOW_INTERVAL).await;
 
-                adc_clone
-                    .usb_host_curr
-                    .topic
-                    .set(adc_clone.usb_host_curr.fast.get());
-                adc_clone
-                    .usb_host1_curr
-                    .topic
-                    .set(adc_clone.usb_host1_curr.fast.get());
-                adc_clone
-                    .usb_host2_curr
-                    .topic
-                    .set(adc_clone.usb_host2_curr.fast.get());
-                adc_clone
-                    .usb_host3_curr
-                    .topic
-                    .set(adc_clone.usb_host3_curr.fast.get());
-                adc_clone
-                    .out0_volt
-                    .topic
-                    .set(adc_clone.out0_volt.fast.get());
-                adc_clone
-                    .out1_volt
-                    .topic
-                    .set(adc_clone.out1_volt.fast.get());
-                adc_clone
-                    .iobus_curr
-                    .topic
-                    .set(adc_clone.iobus_curr.fast.get());
-                adc_clone
-                    .iobus_volt
-                    .topic
-                    .set(adc_clone.iobus_volt.fast.get());
-                adc_clone.pwr_volt.topic.set(adc_clone.pwr_volt.fast.get());
-                adc_clone.pwr_curr.topic.set(adc_clone.pwr_curr.fast.get());
-
-                adc_clone.time.set(Timestamp::now());
+                for channel in &channels {
+                    if let Ok(val) = channel.fast.get() {
+                        // The adc channel topic should likely be wrapped in a Result
+                        // or otherwise be able to contain an error state.
+                        channel.topic.set(val)
+                    }
+                }
+
+                time.set(Timestamp::now());
             }
         });
 

src/adc/iio/demo_mode.rs

@@ -98,18 +98,18 @@ impl CalibratedChannel {
     pub fn try_get_multiple<const N: usize>(
         &self,
         channels: [&Self; N],
-    ) -> Option<[Measurement; N]> {
+    ) -> Result<[Measurement; N]> {
         let ts = Timestamp::now();
         let mut results = [Measurement { ts, value: 0.0 }; N];
 
         for i in 0..N {
-            results[i].value = channels[i].get().value;
+            results[i].value = channels[i].get().unwrap().value;
         }
 
-        Some(results)
+        Ok(results)
     }
 
-    pub fn get(&self) -> Measurement {
+    pub fn get(&self) -> Result<Measurement> {
         let ts = Timestamp::now();
 
         let dt = {
@@ -135,13 +135,13 @@ impl CalibratedChannel {
             .inner
             .parents
             .iter()
-            .map(|p| p.get().value)
+            .map(|p| p.get().unwrap().value)
             .sum::<f32>();
         value += nominal;
 
         self.inner.value.store(value.to_bits(), Ordering::Relaxed);
 
-        Measurement { ts, value }
+        Ok(Measurement { ts, value })
     }
 
     pub fn set(&self, state: bool) {

src/adc/iio/hardware.rs

@@ -27,7 +27,6 @@ use std::time::{Duration, Instant};
 
 use anyhow::{anyhow, Context, Error, Result};
 use async_std::channel::bounded;
-use async_std::stream::StreamExt;
 use async_std::sync::Arc;
 
 use industrial_io::{Buffer, Channel};
@@ -106,8 +105,20 @@ const CHANNELS_PWR: &[ChannelDesc] = &[
 
 const TRIGGER_HR_PWR_DIR: &str = "/sys/kernel/config/iio/triggers/hrtimer/tacd-pwr";
 
+// Timestamps are stored in a 64 Bit atomic variable containing the
+// time in nanoseconds passed since the tacd was started.
+// To reach u64::MAX the tacd would need to run for 2^64ns which is
+// about 584 years.
 const TIMESTAMP_ERROR: u64 = u64::MAX;
 
+#[derive(Debug)]
+pub enum AdcReadError {
+    Again,
+    MismatchedChannels,
+    AquisitionError,
+    TimeStampError,
+}
+
 #[derive(Clone, Copy)]
 struct Calibration {
     scale: f32,
@@ -185,54 +196,56 @@ impl CalibratedChannel {
     pub fn try_get_multiple<const N: usize>(
         &self,
         channels: [&Self; N],
-    ) -> Option<[Measurement; N]> {
+    ) -> Result<[Measurement; N], AdcReadError> {
         let ts_before = self.iio_thread.timestamp.load(Ordering::Acquire);
 
         let mut values_raw = [0; N];
         for (d, ch) in values_raw.iter_mut().zip(channels.iter()) {
-            assert!(
-                Arc::ptr_eq(&self.iio_thread, &ch.iio_thread),
-                "Can only get synchronized adc values for the same thread"
-            );
+            // Can only get time-aligned values for channels of the same ADC
+            if !Arc::ptr_eq(&self.iio_thread, &ch.iio_thread) {
+                return Err(AdcReadError::MismatchedChannels);
+            }
+
             *d = self.iio_thread.values[ch.index].load(Ordering::Relaxed);
         }
 
         let ts_after = self.iio_thread.timestamp.load(Ordering::Acquire);
 
         if ts_before == TIMESTAMP_ERROR || ts_after == TIMESTAMP_ERROR {
-            panic!("Failed to read from ADC");
+            return Err(AdcReadError::AquisitionError);
         }
 
         if ts_before == ts_after {
             let ts = self
                 .iio_thread
                 .ref_instant
                 .checked_add(Duration::from_nanos(ts_before))
-                .unwrap();
+                .ok_or(AdcReadError::TimeStampError)?;
             let ts = Timestamp::new(ts);
 
             let mut values = [Measurement { ts, value: 0.0 }; N];
             for i in 0..N {
                 values[i].value = channels[i].calibration.apply(values_raw[i] as f32);
             }
 
-            Some(values)
+            Ok(values)
         } else {
-            None
+            Err(AdcReadError::Again)
         }
     }
 
     /// Get the value of the channel, or None if the timestamp changed while
     /// reading the value (which should be extremely rare)
-    pub fn try_get(&self) -> Option<Measurement> {
+    pub fn try_get(&self) -> Result<Measurement, AdcReadError> {
         self.try_get_multiple([self]).map(|res| res[0])
     }
 
     // Get the current value of the channel
-    pub fn get(&self) -> Measurement {
+    pub fn get(&self) -> Result<Measurement, AdcReadError> {
         loop {
-            if let Some(r) = self.try_get() {
-                break r;
+            match self.try_get() {
+                Err(AdcReadError::Again) => {}
+                res => break res,
             }
         }
     }
@@ -269,18 +282,23 @@ impl IioThread {
             warn!("Failed to disable {} ADC buffer: {}", adc_name, err);
         }
 
-        let channels: Vec<Channel> = channel_descs
+        let channels: Result<Vec<Channel>> = channel_descs
             .iter()
             .map(|ChannelDesc { kernel_name, .. }| {
                 let ch = adc
                     .find_channel(kernel_name, false)
-                    .unwrap_or_else(|| panic!("Failed to open kernel channel {}", kernel_name));
+                    .ok_or_else(|| anyhow!("Failed to open iio channel {}", kernel_name));
+
+                if let Ok(ch) = ch.as_ref() {
+                    ch.enable();
+                }
 
-                ch.enable();
                 ch
             })
             .collect();
 
+        let channels = channels?;
+
         let trig = ctx
             .find_device(trigger_name)
             .ok_or(anyhow!("Could not find IIO trigger: {}", trigger_name))?;
@@ -292,9 +310,13 @@ impl IioThread {
 
         let buf = adc.create_buffer(buffer_len, false)?;
 
+        let prio = ThreadPriorityValue::try_from(10).map_err(|e| {
+            anyhow!("Failed to set thread priority to 10 as you OS does not support it: {e:?}")
+        })?;
+
         set_thread_priority_and_policy(
             thread_native_id(),
-            ThreadPriority::Crossplatform(ThreadPriorityValue::try_from(10).unwrap()),
+            ThreadPriority::Crossplatform(prio),
             ThreadSchedulePolicy::Realtime(RealtimeThreadSchedulePolicy::Fifo),
         )
         .map_err(|e| anyhow!("Failed to set realtime thread priority: {e:?}"))?;
@@ -317,7 +339,7 @@ impl IioThread {
         // setup was sucessful.
         // This is why we create Self inside the thread and send it back
         // to the calling thread via a queue.
-        let (thread_res_tx, mut thread_res_rx) = bounded(1);
+        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()
@@ -343,6 +365,8 @@ impl IioThread {
                         (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;
                     }
@@ -364,6 +388,8 @@ impl IioThread {
                         // This gives us a chance to return an Err from new().
                         // If the queue was already used just print an error instead.
                         if let Some((_, tx)) = signal_ready.take() {
+                            // Can not fail in practice as the queue is only .take()n
+                            // once and thus known to be empty.
                             tx.try_send(Err(Error::new(e))).unwrap();
                         }
 
@@ -379,24 +405,31 @@ impl IioThread {
                         d.store(s, Ordering::Relaxed)
                     }
 
+                    // These should only fail if
+                    // a) The monotonic time started running backward
+                    // b) The tacd has been running for more than 2**64ns (584 years).
                     let ts: u64 = Instant::now()
                         .checked_duration_since(thread.ref_instant)
-                        .unwrap()
-                        .as_nanos()
-                        .try_into()
-                        .unwrap();
+                        .and_then(|d| d.as_nanos().try_into().ok())
+                        .unwrap_or(TIMESTAMP_ERROR);
 
                     thread.timestamp.store(ts, Ordering::Release);
 
                     // Now that we know that the ADC actually works and we have
                     // initial values: return a handle to it.
                     if let Some((content, tx)) = signal_ready.take() {
+                        // Can not fail in practice as the queue is only .take()n
+                        // once and thus known to be empty.
                         tx.try_send(Ok(content)).unwrap();
                     }
                 }
             })?;
 
-        let thread = thread_res_rx.next().await.unwrap()?;
+        let thread = thread_res_rx.recv().await??;
+
+        // 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);
 
         Ok(thread)
@@ -418,7 +451,7 @@ impl IioThread {
         let hr_trigger_path = Path::new(TRIGGER_HR_PWR_DIR);
 
         if !hr_trigger_path.is_dir() {
-            create_dir(hr_trigger_path).unwrap();
+            create_dir(hr_trigger_path)?;
         }
 
         Self::new("powerboard", "lmp92064", "tacd-pwr", 20, CHANNELS_PWR, 1).await

src/adc/iio/test.rs

@@ -57,7 +57,7 @@ impl CalibratedChannel {
     pub fn try_get_multiple<const N: usize>(
         &self,
         channels: [&Self; N],
-    ) -> Option<[Measurement; N]> {
+    ) -> Result<[Measurement; N]> {
         let mut ts = Timestamp::now();
 
         if self.stall.load(Ordering::Relaxed) {
@@ -78,19 +78,15 @@ impl CalibratedChannel {
             results[i].value = f32::from_bits(val_u32);
         }
 
-        Some(results)
+        Ok(results)
     }
 
-    pub fn try_get(&self) -> Option<Measurement> {
+    pub fn try_get(&self) -> Result<Measurement> {
         self.try_get_multiple([self]).map(|res| res[0])
     }
 
-    pub fn get(&self) -> Measurement {
-        loop {
-            if let Some(r) = self.try_get() {
-                break r;
-            }
-        }
+    pub fn get(&self) -> Result<Measurement> {
+        self.try_get()
     }
 
     pub fn set(&self, val: f32) {

src/dut_power.rs

@@ -353,7 +353,12 @@ impl DutPwrThread {
                             .fast
                             .try_get_multiple([&pwr_volt.fast, &pwr_curr.fast]);
 
-                        if let Some(m) = feedback {
+                        // We do not care too much about _why_ we could not get
+                        // a new value from the ADC.
+                        // If we get a new valid value before the timeout we
+                        // are fine.
+                        // If not we are not.
+                        if let Ok(m) = feedback {
                             last_ts = Some(m[0].ts.as_instant());
                         }
 
@@ -374,7 +379,7 @@ impl DutPwrThread {
                             tick.fetch_add(1, Ordering::Relaxed);
                         }
 
-                        if let Some(m) = feedback {
+                        if let Ok(m) = feedback {
                             break (m[0].value, m[1].value);
                         }
                     };

src/iobus.rs

@@ -142,10 +142,12 @@ impl IoBus {
                 let current = iobus_curr.get();
                 let voltage = iobus_volt.get();
 
-                let undervolt = pwr_en && (voltage.value < VOLTAGE_MIN);
-                let overcurrent = current.value > CURRENT_MAX;
+                if let (Ok(current), Ok(voltage)) = (current, voltage) {
+                    let undervolt = pwr_en && (voltage.value < VOLTAGE_MIN);
+                    let overcurrent = current.value > CURRENT_MAX;
 
-                supply_fault_task.set_if_changed(undervolt || overcurrent);
+                    supply_fault_task.set_if_changed(undervolt || overcurrent);
+                }
 
                 sleep(Duration::from_secs(1)).await;
             }