Apply local settings timeouts, global read_all_settings timeout[CPP-693] (#487)

* Apply local settings timeouts, global read_all_settings timeout[CPP-693]

* Context Receiver Logic -> Watched ConnectionState monitor handle thread. (#491)

Co-authored-by: Steven Meyer <[email protected]>

Author: john-michaelburke

Cargo.lock

@@ -41,7 +41,7 @@ parking_lot = "0.12.0"
 regex = { version = "1.5.5" }
 rust-ini = "0.18.0"
 sbp = { version = "4.1.5", features = ["json", "link", "swiftnav"] }
-sbp-settings = "0.6.7"
+sbp-settings = "0.6.8"
 env_logger = { version = "0.9", optional = true }
 mimalloc = { version = "0.1", default-features = false }
 curl = { version = "0.4", features = ["ssl", "static-curl"] }

console_backend/Cargo.toml

@@ -240,7 +240,9 @@ fn process_messages_thd(
             if let Err(e) = res {
                 error!("Connection error: {}", e);
             }
-            if !matches!(shared_state.connection(), ConnectionState::Disconnected) {
+            if !matches!(shared_state.connection(), ConnectionState::Disconnected)
+                && !shared_state.connection_dialog_visible()
+            {
                 manager_msg.send(ConnectionManagerMsg::Reconnect(conn));
             }
         }

console_backend/src/connection.rs

@@ -2,7 +2,7 @@ use std::borrow::Cow;
 use std::collections::HashMap;
 use std::fs;
 use std::path::Path;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use anyhow::anyhow;
 use capnp::message::Builder;
@@ -32,6 +32,8 @@ const ETHERNET_INTERFACE_MODE_SETTING_KEY: &str = "interface_mode";
 const NTRIP_SETTING_GROUP: &str = "ntrip";
 const NTRIP_ENABLE_SETTING_KEY: &str = "enable";
 
+const SETTINGS_READ_WRITE_TIMEOUT: Duration = Duration::from_millis(1000);
+
 lazy_static! {
     static ref RECOMMENDED_INS_SETTINGS: [(&'static str, &'static str, SettingValue); 4] = [
         ("imu", "imu_raw_output", SettingValue::Boolean(true)),
@@ -334,7 +336,7 @@ impl SettingsTab {
             let value = self
                 .sbp_client
                 .lock()
-                .read_setting(setting.0, setting.1)?
+                .read_setting_with_timeout(setting.0, setting.1, SETTINGS_READ_WRITE_TIMEOUT)?
                 .ok_or_else(|| anyhow!("setting not found"))?
                 .value;
             if value.as_ref() != Some(&setting.2) {
@@ -397,15 +399,20 @@ impl SettingsTab {
                 }
             }
         }
-        let setting = self.sbp_client.lock().write_setting(group, name, value)?;
+        let setting = self.sbp_client.lock().write_setting_with_timeout(
+            group,
+            name,
+            value,
+            SETTINGS_READ_WRITE_TIMEOUT,
+        )?;
         if matches!(
             setting.setting.kind,
             SettingKind::Float | SettingKind::Double
         ) {
             let setting = self
                 .sbp_client
                 .lock()
-                .read_setting(group, name)?
+                .read_setting_with_timeout(group, name, SETTINGS_READ_WRITE_TIMEOUT)?
                 .ok_or_else(|| anyhow!("setting not found"))?;
             self.settings.lock().insert(setting);
         } else {
@@ -419,9 +426,29 @@ impl SettingsTab {
     }
 
     fn read_all_settings(&self) {
-        const TIMEOUT: Duration = Duration::from_secs(15);
+        const GLOBAL_TIMEOUT: Duration = Duration::from_secs(15);
+
+        let (ctx, handle) = Context::with_timeout(SETTINGS_READ_WRITE_TIMEOUT);
+
+        let mut conn = self.shared_state.watch_connection();
+        // this thread runs for at most `GLOBAL_TIMEOUT` seconds so we don't join it to avoid
+        // blocking the completion of this function
+        let _monitor_handle = std::thread::spawn(move || {
+            let t = Instant::now();
+            loop {
+                let waited = t.elapsed();
+                if waited >= GLOBAL_TIMEOUT {
+                    break;
+                }
+                if let Ok(Ok(conn)) = conn.wait_for(GLOBAL_TIMEOUT - waited) {
+                    if !conn.is_connected() {
+                        break;
+                    }
+                }
+            }
+            handle.cancel();
+        });
 
-        let (ctx, _handle) = Context::with_timeout(TIMEOUT);
         let (settings, errors) = self.sbp_client.lock().read_all_ctx(ctx);
         for e in errors {
             warn!("{}", e);

console_backend/src/settings_tab.rs

@@ -4,6 +4,7 @@ use std::{
         atomic::{AtomicBool, AtomicUsize, Ordering},
         Arc,
     },
+    time::{Duration, Instant},
 };
 
 use parking_lot::{Condvar, MappedMutexGuard, Mutex, MutexGuard};
@@ -117,6 +118,14 @@ impl<T: Clone> WatchReceiver<T> {
         Ok(data.value.clone())
     }
 
+    pub fn wait_for(&mut self, duration: Duration) -> Result<Result<T, RecvError>, TimeoutError> {
+        let data = self.wait_for_inner(duration)?;
+        match data {
+            Ok(data) => Ok(Ok(data.value.clone())),
+            Err(e) => Ok(Err(e)),
+        }
+    }
+
     pub fn wait_mut(&mut self) -> Result<MappedMutexGuard<T>, RecvError> {
         let data = self.wait_inner()?;
         Ok(MutexGuard::map(data, |d| &mut d.value))
@@ -155,6 +164,36 @@ impl<T: Clone> WatchReceiver<T> {
         self.last_seen = data.version;
         Ok(data)
     }
+
+    fn wait_for_inner(
+        &mut self,
+        duration: Duration,
+    ) -> Result<Result<MutexGuard<Value<T>>, RecvError>, TimeoutError> {
+        if self.shared.is_closed() {
+            return Ok(Err(RecvError));
+        }
+        let t = Instant::now();
+        let mut data = self.shared.data.lock();
+        while data.version == self.last_seen {
+            let elapsed = t.elapsed();
+            if elapsed >= duration {
+                return Err(TimeoutError);
+            }
+            if self
+                .shared
+                .on_update
+                .wait_for(&mut data, duration - elapsed)
+                .timed_out()
+            {
+                return Err(TimeoutError);
+            }
+            if self.shared.is_closed() {
+                return Ok(Err(RecvError));
+            }
+        }
+        self.last_seen = data.version;
+        Ok(Ok(data))
+    }
 }
 
 impl<T> Clone for WatchReceiver<T> {
@@ -200,6 +239,17 @@ impl fmt::Display for RecvError {
 
 impl std::error::Error for RecvError {}
 
+#[derive(Debug)]
+pub struct TimeoutError;
+
+impl fmt::Display for TimeoutError {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(fmt, "time out waiting for watched value")
+    }
+}
+
+impl std::error::Error for TimeoutError {}
+
 #[cfg(test)]
 mod tests {
     use std::{thread, time::Duration};
@@ -208,6 +258,8 @@ mod tests {
 
     use super::*;
 
+    const SECOND: Duration = Duration::from_secs(1);
+
     #[test]
     fn starts_unseen() {
         let watched = Watched::new(0);
@@ -255,15 +307,15 @@ mod tests {
             s.send(()).unwrap()
         });
 
-        thread::sleep(Duration::from_secs(1));
+        thread::sleep(SECOND);
         assert!(r.try_recv().is_err());
 
         watched.send(1);
-        thread::sleep(Duration::from_secs(1));
+        thread::sleep(SECOND);
         assert!(r.try_recv().is_err());
 
         watched.send(2);
-        thread::sleep(Duration::from_secs(1));
+        thread::sleep(SECOND);
         assert!(r.try_recv().is_ok());
     }
 
@@ -274,12 +326,30 @@ mod tests {
         // mark first as seen
         let _ = recv.get();
         thread::spawn(move || {
-            thread::sleep(Duration::from_secs(1));
+            thread::sleep(SECOND);
             drop(watched);
         });
         assert!(recv.wait().is_err());
     }
 
+    #[test]
+    fn wait_for() {
+        let watched = Watched::new(0);
+        let mut recv = watched.watch();
+        let h = thread::spawn(move || {
+            // mark first as seen
+            let _ = recv.get();
+            let res = recv.wait_for(SECOND / 2);
+            assert!(matches!(res, Err(TimeoutError)));
+            // wait for rest of the second plus a little bit
+            let res = recv.wait_for(SECOND / 2 + Duration::from_millis(100));
+            assert!(matches!(res, Ok(Ok(1))));
+        });
+        thread::sleep(SECOND);
+        watched.send(1);
+        h.join().unwrap();
+    }
+
     #[test]
     fn multiple_consumers() {
         let watched = Watched::new(0);