XcpSendTerminateSessionEvent

build.rs

@@ -29,6 +29,7 @@ fn main() {
         .allowlist_function("XcpEventExt")
         // Misc
         .allowlist_function("XcpPrint")
+        .allowlist_function("XcpSendTerminateSessionEvent")
         .allowlist_function("ApplXcpSetLogLevel")
         .allowlist_function("ApplXcpSetA2lName")
         .allowlist_function("ApplXcpSetEpk")

examples/xcp_daemon/Cargo.toml

@@ -10,7 +10,7 @@ log = "0.4.22"
 
 # In unix we link our dependencies as per usual
 [target.'cfg(unix)'.dependencies]
-xcp = { path = "../../", features = [] }
+xcp = { path = "../../", features = ["xcp_daemon"] }
 
 log = "0.4.22"
 signal-hook = "0.3.17"

src/reg/registry.rs

@@ -687,6 +687,7 @@ impl Registry {
 
     // Add a calibration segment
     pub fn add_cal_seg(&mut self, name: &'static str, index: u16, size: u32) {
+        // Panic if registry is closed
         assert!(!self.is_frozen(), "Registry is closed");
 
         // Length of calseg should be %4 to avoid problems with CANape and checksum calculations
@@ -826,7 +827,7 @@ impl Registry {
         }
 
         // Sort measurement and calibration lists to get deterministic order
-        // Event and CalSeg lists stay in the order the were added
+        // Event and CalSeg lists stay in the order they were added
         self.measurement_list.sort();
         self.characteristic_list.sort();
 

src/xcp.rs

@@ -166,7 +166,7 @@ impl XcpEvent {
         // @@@@ Unsafe - C library call and transfering a pointer and its valid memory range to XCPlite FFI
         #[cfg(not(feature = "xcp_appl"))]
         unsafe {
-            xcplib::XcpEvent(self.get_channel())
+            xcplib::XcpEvent(self.get_channel());
         }
         #[cfg(feature = "xcp_appl")]
         unsafe {
@@ -477,10 +477,10 @@ impl Xcp {
 
     #[allow(clippy::unused_self)]
     /// Set the log level for XCP library
-    pub fn set_log_level(&self, _level: u8) {
+    pub fn set_log_level(&self, level: u8) {
         unsafe {
             // @@@@ Unsafe - C library call
-            xcplib::ApplXcpSetLogLevel(_level);
+            xcplib::ApplXcpSetLogLevel(level);
         }
     }
 
@@ -509,14 +509,23 @@ impl Xcp {
     /// Stop the XCP server
     #[allow(clippy::unused_self)]
     pub fn stop_server(&self) {
+        // @@@@ Unsafe - C library calls
         unsafe {
-            // @@@@ Unsafe - C library call
+            xcplib::XcpSendTerminateSessionEvent(); // Send terminate session event, if the XCP client is still connected
             xcplib::XcpDisconnect();
-            // @@@@ Unsafe - C library call
             xcplib::XcpEthServerShutdown();
         }
     }
 
+    /// Signal the client to disconnect
+    #[allow(clippy::unused_self)]
+    pub fn disconnect_client(&self) {
+        // @@@@ Unsafe - C library calls
+        unsafe {
+            xcplib::XcpSendTerminateSessionEvent(); // Send terminate session event, if the XCP client is connected
+        }
+    }
+
     //------------------------------------------------------------------------------------------
     // Calibration segments
 

src/xcp/cal/cal_seg.rs

@@ -141,7 +141,7 @@ impl<T> CalPageTrait for T where T: Sized + Send + Sync + Copy + Clone + 'static
 /// Each instance stores 2 copies of its inner data, the calibration page
 /// One for each clone of the readers, a shared copy for the writer (XCP) and
 /// a reference to the default values
-/// Implements Deref to simplify usage
+/// Implements Deref to simplify usage, is send, not sync and implements copy and clone
 ///
 
 #[derive(Debug)]
@@ -153,8 +153,7 @@ where
     default_page: &'static T,
     ecu_page: Box<CalPage<T>>,
     xcp_page: Arc<Mutex<CalPage<T>>>,
-    //_not_send_sync_marker: PhantomData<*mut ()>,
-    _not_sync_marker: PhantomData<std::cell::Cell<()>>,
+    _not_sync_marker: PhantomData<std::cell::Cell<()>>, // CalSeg is send, not sync
 }
 
 // Impl register_fields for types which implement RegisterFieldsTrait
@@ -381,13 +380,13 @@ where
     // Read from xcp_page or default_page depending on the active XCP page
     // # Safety
     // dst must be valid
-    // @@@@ Unsafe function
+    // @@@@ - Unsafe function
     unsafe fn read(&self, offset: u16, len: u8, dst: *mut u8) -> bool;
 
     // Write to xcp_page
     // # Safety
     // src must be valid
-    // @@@@ Unsafe function
+    // @@@@ - Unsafe function
     unsafe fn write(&self, offset: u16, len: u8, src: *const u8, delay: u8) -> bool;
 
     // Flush delayed modifications
@@ -417,7 +416,7 @@ where
         self.xcp_page.lock().init_request = true;
     }
 
-    // @@@@ Unsafe
+    // @@@@ Unsafe - function
     unsafe fn read(&self, offset: u16, len: u8, dst: *mut u8) -> bool {
         assert!(offset as usize + len as usize <= std::mem::size_of::<T>());
         if Xcp::get().get_xcp_cal_page() == XcpCalPage::Ram {
@@ -432,7 +431,7 @@ where
         }
     }
 
-    // @@@@ Unsafe
+    // @@@@ Unsafe - function
     unsafe fn write(&self, offset: u16, len: u8, src: *const u8, delay: u8) -> bool {
         assert!(offset as usize + len as usize <= std::mem::size_of::<T>());
         if Xcp::get().get_xcp_cal_page() == XcpCalPage::Ram {
@@ -533,19 +532,7 @@ where
 // }
 
 //----------------------------------------------------------------------------------------------
-// Send marker
-
-// The Send marker trait indicates that ownership of the type can be transferred to a different thread.
-// The Sync marker trait would indicates that it is safe to share references to CalSeg between threads, which is not the case.
-
-/// Send marker for 'CalSeg'
-/// 'CalSeg' is not Sync, but Send
-/// # Safety
-/// This is safe, because 'CalSeg' would be Send and Sync, but its disabled by PhantomData
-/// 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: CalPageTrait {}
+// Read lock guard for calibration pages
 
 /// Read lock guard that provides consistent read only access to a calibration page
 pub struct ReadLockGuard<'a, T: CalPageTrait> {

src/xcp/xcplib.rs

@@ -220,6 +220,9 @@ extern "C" {
 extern "C" {
     pub fn XcpEvent(event: u16);
 }
+extern "C" {
+    pub fn XcpSendTerminateSessionEvent();
+}
 extern "C" {
     pub fn XcpPrint(str_: *const ::std::os::raw::c_char);
 }

tests/test_single_thread.rs

@@ -147,6 +147,8 @@ fn task(cal_seg: CalSeg<CalPage1>) {
     }
 
     debug!("Task terminated, loop counter = {}, {} changes observed", loop_counter, changes);
+    xcp_println!("Task terminated, loop counter = {}, {} changes observed", loop_counter, changes);
+    Xcp::disconnect_client(Xcp::get());
 }
 
 //-----------------------------------------------------------------------------

tests/xcp_test_executor.rs

@@ -481,7 +481,7 @@ pub async fn xcp_test_executor(_xcp: &Xcp, test_mode_cal: TestModeCal, test_mode
                 assert_eq!(d.counter_errors, 0);
                 assert_eq!(d.packets_lost, 0);
             }
-        }
+        } // test_mode_daq == TestModeDaq::SingleThreadDAQ || test_mode_daq == TestModeDaq::MultiThreadDAQ
 
         //-------------------------------------------------------------------------------------------------------------------------------------
         //-------------------------------------------------------------------------------------------------------------------------------------
@@ -602,12 +602,12 @@ pub async fn xcp_test_executor(_xcp: &Xcp, test_mode_cal: TestModeCal, test_mode
                 if download_time > 100.0 {
                     warn!("Calibration download time ({}us) is too high!", download_time);
                 }
-            }
+            } // Calibration test loop
 
+            /* @@@@ TODO reenable this
             // Consistent calibration test loop
             // Do MAX_ITER consistent calibrations on cal_seg.sync_test1/2 cal_test, task will panic if different
             warn!("Consistent calibration test disabled");
-            /* @@@@ TODO reenable this
             {
                 tokio::time::sleep(Duration::from_micros(10000)).await;
 
@@ -669,9 +669,9 @@ pub async fn xcp_test_executor(_xcp: &Xcp, test_mode_cal: TestModeCal, test_mode
 
                     info!("consistent calibration test loop done, {} iterations", CAL_TEST_MAX_ITER);
                 }
-            }
+            } // Consistent calibration test loop
             */
-        }
+        } // !error_state && (test_mode_cal == TestModeCal::Cal)
 
         // Stop test task
         info!("Stop test tasks");
@@ -684,7 +684,8 @@ pub async fn xcp_test_executor(_xcp: &Xcp, test_mode_cal: TestModeCal, test_mode
             })
             .ok();
 
-        tokio::time::sleep(Duration::from_millis(500)).await; // Give the user task some time to finish
+        info!("...");
+        tokio::time::sleep(Duration::from_millis(1000)).await; // Give the user task some time to finish
     }
 
     // Disconnect

xcp_client/src/xcp_client.rs

@@ -19,6 +19,7 @@ use std::io::Write;
 use std::net::SocketAddr;
 use std::path::Path;
 use std::sync::Arc;
+use tokio::io::join;
 use tokio::net::UdpSocket;
 use tokio::select;
 use tokio::sync::mpsc::{self, Receiver, Sender};
@@ -70,6 +71,8 @@ pub const ERROR_TL_HEADER: u8 = 0xF1;
 pub const ERROR_A2L: u8 = 0xF2;
 pub const ERROR_LIMIT: u8 = 0xF3;
 pub const ERROR_ODT_SIZE: u8 = 0xF4;
+pub const ERROR_TASK_TERMINATED: u8 = 0xF5;
+pub const ERROR_SESSION_TERMINATION: u8 = 0xF6;
 
 #[derive(Default)]
 pub struct XcpError {
@@ -93,6 +96,12 @@ impl std::fmt::Display for XcpError {
             ERROR_CMD_TIMEOUT => {
                 write!(f, "{cmd:?}: Command response timeout")
             }
+            ERROR_TASK_TERMINATED => {
+                write!(f, "Client task terminated")
+            }
+            ERROR_SESSION_TERMINATION => {
+                write!(f, "Session terminated by XCP server")
+            }
             ERROR_TL_HEADER => {
                 write!(f, "Transport layer header error")
             }
@@ -546,6 +555,7 @@ pub struct XcpClient {
     bind_addr: SocketAddr,
     dest_addr: SocketAddr,
     socket: Option<Arc<UdpSocket>>,
+    receive_task: Option<tokio::task::JoinHandle<()>>,
     rx_cmd_resp: Option<mpsc::Receiver<Vec<u8>>>,
     tx_task_control: Option<mpsc::Sender<XcpTaskControl>>,
     task_control: XcpTaskControl,
@@ -570,6 +580,7 @@ impl XcpClient {
             bind_addr,
             dest_addr,
             socket: None,
+            receive_task: None,
             rx_cmd_resp: None,
             tx_task_control: None,
             task_control: XcpTaskControl::new(),
@@ -641,7 +652,7 @@ impl XcpClient {
                         Ok((size, _)) => {
                             // Handle the data from recv_from
                             if size == 0 {
-                                warn!("xcp_receive: socket closed");
+                                warn!("receive_task: stop, socket closed");
                                 return Ok(());
                             }
 
@@ -669,19 +680,23 @@ impl XcpClient {
                                     0xFF => {
                                         // Command response
                                         let response = &buf[(i + 4)..(i + 4 + len)];
-                                        trace!("xcp_receive: XCP response = {:?}", response);
+                                        trace!("receive_task: XCP response = {:?}", response);
                                         tx_resp.send(response.to_vec()).await?;
                                     }
                                     0xFE => {
                                         // Command error response
                                         let response = &buf[(i + 4)..(i + 6)];
-                                        trace!("xcp_receive: XCP error response = {:?}", response);
+                                        trace!("receive_task: XCP error response = {:?}", response);
                                         tx_resp.send(response.to_vec()).await?;
                                     }
                                     0xFD => {
                                         // Event
                                         let event_code = buf[i + 5];
-                                        warn!("xcp_receive: ignored XCP event = 0x{:0X}", event_code);
+                                        match event_code {
+                                            0x07 => { warn!("receive_task: stop, SESSION_TERMINATDED"); return Err(Box::new(XcpError::new(ERROR_SESSION_TERMINATION,0)) as Box<dyn Error>); },
+                                            _ => warn!("xcp_receive: ignored XCP event = 0x{:0X}", event_code),
+                                        }
+
                                     }
                                     0xFC => {
                                         // Service
@@ -691,7 +706,7 @@ impl XcpClient {
                                         } else {
                                             // Unknown PID
                                             warn!(
-                                                "xcp_receive: ignored unknown service request code = 0x{:0X}",
+                                                "receive_task: ignored unknown service request code = 0x{:0X}",
                                                 service_code
                                             );
                                         }
@@ -716,7 +731,7 @@ impl XcpClient {
                         }
                         Err(e) => {
                             // Handle the error from recv_from
-                            error!("xcp_receive: socket error {}",e);
+                            warn!("receive_task: stop, socket error {}",e);
                             return Err(Box::new(XcpError::new(ERROR_TL_HEADER,0)) as Box<dyn Error>);
                         }
                     }
@@ -756,9 +771,9 @@ impl XcpClient {
                         }
                     }
                     None => {
-                        // @@@@ Empty response, channel has been closed, return with XcpError Timeout
-                        error!("xcp_command: receive_task channel closed");
-                        Err(Box::new(XcpError::new(ERROR_CMD_TIMEOUT, 0)) as Box<dyn Error>)
+                        // Empty response, channel has been closed because receive task terminated
+                        warn!("xcp_command: receive_task terminated");
+                        Err(Box::new(XcpError::new(ERROR_TASK_TERMINATED, cmd_bytes[4])) as Box<dyn Error>)
                     }
                 }
             }
@@ -792,9 +807,9 @@ impl XcpClient {
             self.tx_task_control = Some(tx_daq); // tx XCP DAQ control channel
             let daq_decoder_clone = Arc::clone(&daq_decoder);
 
-            tokio::spawn(async move {
+            self.receive_task = Some(tokio::spawn(async move {
                 let _res = XcpClient::receive_task(socket, tx_resp, rx_daq, text_decoder, daq_decoder_clone).await;
-            });
+            }));
             tokio::time::sleep(Duration::from_millis(100)).await; // wait for the receive task to start
         }
 
@@ -832,11 +847,23 @@ impl XcpClient {
 
     //------------------------------------------------------------------------
     pub async fn disconnect(&mut self) -> Result<(), Box<dyn Error>> {
-        self.send_command(XcpCommandBuilder::new(CC_DISCONNECT).add_u8(0).build()).await?;
+        // Ignore errors and assume disconnected
+
+        // Disconnect
+        let _ = self.send_command(XcpCommandBuilder::new(CC_DISCONNECT).add_u8(0).build()).await;
 
+        // Stop XCP client task
         self.task_control.connected = false;
         self.task_control.running = false;
-        self.tx_task_control.as_ref().unwrap().send(self.task_control).await.unwrap();
+        let _ = self.tx_task_control.as_ref().unwrap().send(self.task_control).await;
+
+        // Make sure receive_task has terminated
+        if let Some(receive_task) = self.receive_task.take() {
+            let res = receive_task.await;
+            if let Err(e) = res {
+                error!("{:?}", e);
+            }
+        }
 
         Ok(())
     }

xcp_lite.a2l

@@ -64,7 +64,7 @@
 
 
 		/begin MOD_PAR ""
-			EPK "2025-01-06 18:50:11Z" ADDR_EPK 0x80000000
+			EPK "2025-01-09 17:28:40Z" ADDR_EPK 0x80000000
 
 			/begin MEMORY_SEGMENT epk "" DATA FLASH INTERN 0x80000000 20 -1 -1 -1 -1 -1 /end MEMORY_SEGMENT
 
@@ -161,7 +161,7 @@
 
 			/end DAQ
 
-			/begin XCP_ON_UDP_IP 0x104 5555 ADDRESS "192.168.0.128" /end XCP_ON_UDP_IP
+			/begin XCP_ON_UDP_IP 0x104 5555 ADDRESS "192.168.0.83" /end XCP_ON_UDP_IP
 
 		/end IF_DATA