xcp_client improvements and clock test

Author: RainerZ

Cargo.toml

@@ -90,7 +90,8 @@ thiserror = "1.0.64"
 # Command line parser
 clap = { version = "4.5.9", features = ["derive"] }
 
-# Raw FFI bindings to platform libraries like libc
+# Raw FFI bindings to platform libraries
+# For XcpLite
 libc = "0.2.153"
 
 # A macro to generate structures which behave like bitflags
@@ -145,7 +146,7 @@ tokio = { version = "1.37.0", features = ["full"] }
 a2lfile = { version="2.2.0", optional = false}
 xcp_client = { path = "xcp_client" }
 
-# dependencies for rayon demo example
+# dependencies for point_cloud example
 cdr = "0.2.4"
 
 # dependencies for rayon demo example

benches/xcp_benchmark.rs

@@ -131,6 +131,7 @@ async fn xcp_client(dest_addr: std::net::SocketAddr, local_addr: std::net::Socke
                     xcp_client.create_measurement_object("signal7").expect("measurement signal not found");
                     xcp_client.create_measurement_object("signal8").expect("measurement signal not found");
                     // Measure start
+                    xcp_client.init_measurement().await.unwrap();
                     xcp_client.start_measurement().await.expect("could not start measurement");
                 }
 

src/xcp/cal/cal_seg.rs

@@ -109,7 +109,7 @@ macro_rules! calseg_field {
 // Calibration parameter page wrapper for T with modification counter, init and freeze requests
 
 #[derive(Debug, Copy, Clone)]
-struct CalPage<T: Sized + Send + Sync + Copy + Clone + 'static> {
+struct CalPage<T: CalPageTrait> {
     ctr: u16,
     init_request: bool,
     freeze_request: bool,
@@ -155,7 +155,7 @@ impl<T> CalPageTrait for T where T: Sized + Send + Sync + Copy + Clone + 'static
 #[derive(Debug)]
 pub struct CalSeg<T>
 where
-    T: Sized + Send + Sync + Copy + Clone + 'static,
+    T: CalPageTrait,
 {
     index: usize,
     default_page: &'static T,
@@ -167,7 +167,7 @@ where
 // Impl register_fields for types which implement RegisterFieldsTrait
 impl<T> CalSeg<T>
 where
-    T: RegisterFieldsTrait,
+    T: CalPageTrait + RegisterFieldsTrait,
 {
     /// Register all fields of a calibration segment in the registry
     /// Requires the calibration page to implement XcpTypeDescription
@@ -181,7 +181,7 @@ where
 #[cfg(feature = "serde")]
 impl<T> CalSeg<T>
 where
-    T: Sized + Send + Sync + Copy + Clone + 'static + serde::Serialize + serde::de::DeserializeOwned,
+    T: CalPageTrait,
 {
     /// Load a calibration segment from json file
     /// Requires the calibration page type to implement serde::Serialize + serde::de::DeserializeOwned
@@ -313,7 +313,7 @@ where
 
                 let file = std::fs::File::create(path).unwrap();
                 let mut writer = std::io::BufWriter::new(file);
-                let s = serde_json::to_string(&self.xcp_page.lock().page)
+                let s = serde_json::to_string(&xcp_page.page)
                     .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, format!("serde_json::to_string failed: {}", e)))
                     .unwrap();
                 std::io::Write::write_all(&mut writer, s.as_ref()).unwrap();
@@ -399,7 +399,7 @@ where
 
 impl<T> CalSegTrait for CalSeg<T>
 where
-    T: Sized + Send + Sync + Copy + Clone + 'static,
+    T: CalPageTrait,
 {
     fn get_name(&self) -> &'static str {
         Xcp::get().get_calseg_name(self.index)
@@ -463,7 +463,7 @@ where
 
 impl<T> Deref for CalSeg<T>
 where
-    T: Sized + Send + Sync + Copy + Clone + 'static,
+    T: CalPageTrait,
 {
     type Target = T;
 
@@ -484,7 +484,7 @@ where
 // This is undefined behaviour, because the reference to XCP data page will escape from its mutex
 impl<T> std::ops::DerefMut for CalSeg<T>
 where
-    T: Sized + Send + Sync + Copy + Clone + 'static,
+    T: CalPageTrait,
 {
     fn deref_mut(&mut self) -> &mut Self::Target {
         warn!("Unsafe deref mut to XCP page of {}, this is undefined behaviour !!", self.get_name());
@@ -500,7 +500,7 @@ where
 
 impl<T> Clone for CalSeg<T>
 where
-    T: Sized + Send + Sync + Copy + Clone + 'static,
+    T: CalPageTrait,
 {
     fn clone(&self) -> Self {
         CalSeg {
@@ -519,7 +519,7 @@ where
 
 // impl<T> Drop for CalSeg<T>
 // where
-//     T: Sized + Send + Sync + Copy + Clone + 'static,
+//     T: CalPageTrait,
 // {
 //     fn drop(&mut self) {
 //         let clone_count = self.get_clone_count();
@@ -546,7 +546,7 @@ where
 /// Send is reimplemented here
 /// Sync stays disabled, because this would allow to call calseg.sync() from multiple threads with references to the same CalSeg
 // @@@@ unsafe - Implementation of Send marker for CalSeg
-unsafe impl<T> Send for CalSeg<T> where T: Sized + Send + Sync + Copy + Clone + 'static {}
+unsafe impl<T> Send for CalSeg<T> where T: CalPageTrait {}
 
 //----------------------------------------------------------------------------------------------
 // Test
@@ -592,11 +592,13 @@ mod cal_tests {
         Ok(())
     }
 
+    #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
     #[derive(Debug, Clone, Copy)]
     struct CalPage0 {
         stop: u8,
     }
 
+    #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
     #[derive(Debug, Clone, Copy, XcpTypeDescription)]
     struct CalPageTest {
         test: u8,
@@ -774,7 +776,7 @@ mod cal_tests {
         save(&mut_page, "test_cal_seg.json").unwrap();
 
         // Create a cal_seg with a mut_page from file test_cal_seg.json aka CAL_PAR_RAM, and a default page from CAL_PAR_FLASH
-        let mut cal_seg = xcp.create_calseg("test_cal_seg", &CAL_PAR_FLASH);
+        let cal_seg = xcp.create_calseg("test_cal_seg", &CAL_PAR_FLASH);
         cal_seg.load("test_cal_seg.json").unwrap();
 
         let cal_seg1 = cal_seg.clone();
@@ -855,7 +857,7 @@ mod cal_tests {
         let mut_page: CalPage2 = CalPage2 { a: 1, b: 3, c: 5 };
         save(&mut_page, "test1.json").unwrap();
         save(&mut_page, "test2.json").unwrap();
-        let mut cal_seg = xcp.create_calseg("test1", &FLASH_PAGE2);
+        let cal_seg = xcp.create_calseg("test1", &FLASH_PAGE2);
         cal_seg.load("test1.json").unwrap();
         cal_seg.sync();
         assert_eq!(xcp.get_ecu_cal_page(), XcpCalPage::Ram, "XCP should be on RAM page here, there is no independant page switching yet");
@@ -886,11 +888,10 @@ mod cal_tests {
 
     //-----------------------------------------------------------------------------
     // Test cal page freeze
-
     #[cfg(feature = "serde")]
     #[test]
     fn test_cal_page_freeze() {
-        let xcp = xcp_test::test_setup(log::LevelFilter::Warn);
+        let xcp = xcp_test::test_setup(log::LevelFilter::Info);
 
         assert!(std::mem::size_of::<CalPage1>() == 12);
         assert!(std::mem::size_of::<CalPage2>() == 12);
@@ -900,7 +901,7 @@ mod cal_tests {
         save(&mut_page1, "test1.json").unwrap();
 
         // Create calseg1 from def
-        let mut calseg1 = xcp.create_calseg("test1", &FLASH_PAGE1);
+        let calseg1 = xcp.create_calseg("test1", &FLASH_PAGE1);
         calseg1.load("test1.json").unwrap();
 
         test_is_mut!(calseg1);
@@ -912,7 +913,7 @@ mod cal_tests {
 
         // Create calseg2 from freeze file test1.json of calseg1
         std::fs::copy("test1.json", "test2.json").unwrap();
-        let mut calseg2 = xcp.create_calseg("test2", &FLASH_PAGE2);
+        let calseg2 = xcp.create_calseg("test2", &FLASH_PAGE2);
         calseg2.load("test2.json").unwrap();
 
         test_is_mut!(calseg2);

tests/test_tokio_multi_thread.rs

@@ -276,7 +276,7 @@ async fn task(index: usize, cal_seg: CalSeg<CalPage1>) {
 //-----------------------------------------------------------------------------
 // Integration test multi thread measurememt and calibration
 
-#[ignore]
+//#[ignore]
 #[tokio::test]
 async fn test_tokio_multi_thread() {
     env_logger::Builder::new()

tests/test_tokio_single_thread.rs

@@ -145,7 +145,7 @@ fn task(cal_seg: CalSeg<CalPage1>) {
 //-----------------------------------------------------------------------------
 // Integration test single thread measurement and calibration
 
-#[ignore]
+//#[ignore]
 #[tokio::test]
 async fn test_tokio_single_thread() {
     env_logger::Builder::new()

tests/xcp_test_executor.rs

@@ -64,6 +64,7 @@ impl XcpTextDecoder for ServTextDecoder {
 
 #[derive(Debug, Clone, Copy)]
 struct DaqDecoder {
+    timestamp_resolution: u64,
     tot_events: u32,
     packets_lost: u32,
     counter_errors: u32,
@@ -78,6 +79,7 @@ struct DaqDecoder {
 impl DaqDecoder {
     pub fn new() -> DaqDecoder {
         DaqDecoder {
+            timestamp_resolution: 1,
             tot_events: 0,
             packets_lost: 0,
             counter_errors: 0,
@@ -107,6 +109,11 @@ impl XcpDaqDecoder for DaqDecoder {
         }
     }
 
+    // Set timestamp resolution
+    fn set_timestamp_resolution(&mut self, timestamp_resolution: u64) {
+        self.timestamp_resolution = timestamp_resolution;
+    }
+
     // Handle incomming DAQ DTOs from XCP server
     fn decode(&mut self, lost: u32, daq: u16, odt: u8, timestamp: u32, data: &[u8]) {
         assert!(daq < MULTI_THREAD_TASK_COUNT as u16);
@@ -120,7 +127,7 @@ impl XcpDaqDecoder for DaqDecoder {
             self.daq_max = daq;
         }
 
-        // Decode time as u64
+        // Decode raw timestamp as u64
         // Check declining timestamps
         if odt == 0 {
             let t_last = self.daq_timestamp[daq as usize];
@@ -325,7 +332,6 @@ pub async fn xcp_test_executor(xcp: &Xcp, test_mode_cal: TestModeCal, test_mode_
         // DAQ test single_thread or multi_thread
         if test_mode_daq == TestModeDaq::SingleThreadDAQ || test_mode_daq == TestModeDaq::MultiThreadDAQ {
             tokio::time::sleep(Duration::from_micros(10000)).await;
-            info!("Start data acquisition test");
 
             // Create a calibration object for CalPage1.counter_max
             // Set counter_max to 15
@@ -339,6 +345,21 @@ pub async fn xcp_test_executor(xcp: &Xcp, test_mode_cal: TestModeCal, test_mode_
             // Set cycle time
             xcp_client.set_value_u64(cycle_time_us, DAQ_TEST_TASK_SLEEP_TIME_US).await.unwrap();
 
+            // Check the DAQ clock
+            info!("Start clock test");
+            let t10 = Instant::now();
+            let t1 = xcp_client.get_daq_clock().await.unwrap();
+            tokio::time::sleep(Duration::from_micros(1000)).await;
+            let t20 = t10.elapsed();
+            let t2 = xcp_client.get_daq_clock().await.unwrap();
+            let dt12 = (t2 - t1) / 1000;
+            let dt120 = t20.as_micros() as u64;
+            info!("t1 = {}ns, t2 = {}ns, dt={}us / elapsed={}us", t1, t2, dt12, dt120);
+            assert!(dt12 > dt120 - 100, "DAQ clock too slow");
+            assert!(dt12 < dt120 + 100, "DAQ clock too fast");
+
+            info!("Start data acquisition test");
+
             // Measurement test loop
             // Create a measurement DAQ list with all instances MULTI_THREAD_TASK_COUNT of measurement counter and counter_max
             // Hard coded order and size in DaqDecoder (counter_max, counter, cal_test, ...)

xcp_client/README.md

@@ -33,9 +33,10 @@ Using tokio and a2lfile.
 
     // Measurement
     // Create a measurement for signal counter:u32
+    xcp_client.init_measurement().await?;
     xcp_client.create_measurement_object("counter").await?;
     xcp_client.start_measurement().await?;
-    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+    sleep(Duration::from_secs(1)).await;
     xcp_client.stop_measurement().await?;
 
     // Disconnect

xcp_client/src/main.rs

@@ -40,6 +40,7 @@ const MAX_EVENT: usize = 16;
 
 #[derive(Debug)]
 struct DaqDecoder {
+    timestamp_resolution: u64,
     event_count: usize,
     byte_count: usize,
     daq_timestamp: [u64; MAX_EVENT],
@@ -48,6 +49,7 @@ struct DaqDecoder {
 impl DaqDecoder {
     pub fn new() -> DaqDecoder {
         DaqDecoder {
+            timestamp_resolution: 0,
             event_count: 0,
             byte_count: 0,
             daq_timestamp: [0; MAX_EVENT],
@@ -68,6 +70,11 @@ impl XcpDaqDecoder for DaqDecoder {
         }
     }
 
+    // Set timestamp resolution
+    fn set_timestamp_resolution(&mut self, timestamp_resolution: u64) {
+        self.timestamp_resolution = timestamp_resolution;
+    }
+
     // Decode DAQ data
     fn decode(&mut self, lost: u32, daq: u16, odt: u8, timestamp: u32, data: &[u8]) {
         assert!(daq < MAX_EVENT as u16);
@@ -93,7 +100,8 @@ impl XcpDaqDecoder for DaqDecoder {
             assert!(data.len() >= 4);
             let counter = data[0] as u32 | (data[1] as u32) << 8 | (data[2] as u32) << 16 | (data[3] as u32) << 24;
             //trace!("DAQ: lost={}, daq={}, odt={}: timestamp={} counter={} data={:?}", lost, daq, odt, t, counter, data);
-            info!("DAQ: lost={}, daq={}, odt={}, t={}, counter={}", lost, daq, odt, t, counter);
+            let t = t * self.timestamp_resolution;
+            info!("DAQ: lost={}, daq={}, odt={}, t={}ns, counter={}", lost, daq, odt, t, counter);
         }
 
         self.byte_count += data.len(); // overall payload byte count