rewrote ethercrab and added one cycle version instead of blocking

Cargo.lock

@@ -1036,8 +1036,7 @@ dependencies = [
 [[package]]
 name = "ethercrab"
 version = "0.6.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fbb13becfa15713adb09b27fdb74468b827cdc78aa99aea638a89c1b0db3ce94"
+source = "git+https://github.com/kraemr/ethercrab-deterministic.git?rev=53533ee42d11a32f6b8638233b59c402f3eb1998#53533ee42d11a32f6b8638233b59c402f3eb1998"
 dependencies = [
  "async-io",
  "atomic-waker",
@@ -1066,8 +1065,7 @@ dependencies = [
 [[package]]
 name = "ethercrab-wire"
 version = "0.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d15828cb020fb44c92ca67e1ee3c122c9621c2e00d383d066e488f0619023ba2"
+source = "git+https://github.com/kraemr/ethercrab-deterministic.git?rev=53533ee42d11a32f6b8638233b59c402f3eb1998#53533ee42d11a32f6b8638233b59c402f3eb1998"
 dependencies = [
  "ethercrab-wire-derive",
  "heapless",
@@ -1076,8 +1074,7 @@ dependencies = [
 [[package]]
 name = "ethercrab-wire-derive"
 version = "0.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0e790b8011211ed70babfcacbd4277b91ee21402ad5ba6a2c293db02e2f12919"
+source = "git+https://github.com/kraemr/ethercrab-deterministic.git?rev=53533ee42d11a32f6b8638233b59c402f3eb1998#53533ee42d11a32f6b8638233b59c402f3eb1998"
 dependencies = [
  "criterion",
  "proc-macro2",

Cargo.toml

@@ -25,3 +25,6 @@ style = { level = "warn", priority = -1 }
 pendantic = { level = "warn", priority = -1 }
 cargo = { level = "warn", priority = -1 }
 nursery = { level = "warn", priority = -1 }
+
+[patch.crates-io]
+ethercrab = { git = "https://github.com/kraemr/ethercrab-deterministic.git",rev = "53533ee42d11a32f6b8638233b59c402f3eb1998"  }

server/src/ethercat/setup.rs

@@ -17,40 +17,62 @@ use control_core::machines::new::MachineNewHardwareEthercat;
 use control_core::realtime::{set_core_affinity, set_realtime_priority};
 use control_core::socketio::namespace::NamespaceCacheingLogic;
 use ethercat_hal::devices::devices_from_subdevices;
-use ethercrab::std::ethercat_now;
-use ethercrab::{MainDevice, MainDeviceConfig, PduStorage, RetryBehaviour, Timeouts};
+use ethercrab::std::{ethercat_now, tx_rx_task};
+use ethercrab::{
+    MainDevice, MainDeviceConfig, PduLoop, PduRx, PduStorage, PduTx, RetryBehaviour, Timeouts,
+};
 use std::{sync::Arc, time::Duration};
 
 const SM_OUTPUT: u16 = 0x1C32;
 const SM_INPUT: u16 = 0x1C33;
 
-pub async fn setup_loop(interface: &str, app_state: Arc<AppState>) -> Result<(), anyhow::Error> {
-    tracing::info!("Starting Ethercat PDU loop");
+pub struct EtherCatBackend<'maindevice> {
+    pub tx: PduTx<'maindevice>,
+    pub rx: PduRx<'maindevice>,
+    pub interface: String,
+}
 
-    // Erase all all setup data from `app_state`
-    {
-        tracing::debug!("Setting up Ethercat network");
-        let mut ethercat_setup_guard = app_state.ethercat_setup.write().await;
-        *ethercat_setup_guard = None;
-    }
+/*
+pub async hack_el5125(){
+        for subdevice in subdevices.iter() {
+            if subdevice.name() == "EL5152" {
+                subdevice.sdo_write(SM_OUTPUT, 0x1, 0x00u16).await; //set sync mode (1) for free run (0)
+                subdevice.sdo_write(SM_INPUT, 0x1, 0x00u16).await; //set sync mode (1) for free run (0)
+            }
+        }
+}
+*/
+pub async fn setup_loop(
+    interface: &str,
+    app_state: Arc<AppState>,
+) -> Result<EtherCatBackend<'_>, anyhow::Error> {
+    tracing::info!("Starting EtherCAT PDU loop");
 
-    // Setup ethercrab tx/rx task
+    // Set real-time priority and CPU affinity
+    set_realtime_priority();
+    #[cfg(all(target_os = "linux", not(feature = "development-build")))]
+    set_irq_affinity(interface, 3)
+        .map(|_| tracing::info!("Ethernet IRQ on CPU 3"))
+        .map_err(|e| tracing::error!("Failed IRQ affinity: {:?}", e))
+        .ok();
+
+    *app_state.ethercat_setup.write().await = None;
+
+    // Setup PDU storage
     let pdu_storage = Box::leak(Box::new(PduStorage::<MAX_FRAMES, MAX_PDU_DATA>::new()));
     let (tx, rx, pdu) = pdu_storage.try_split().expect("can only split once");
-    let interface = interface.to_string();
-
-    std::thread::Builder::new()
+    let interface: String = interface.to_string();
+    let cloned_interface = interface.clone();
+    let ethercat_thread = std::thread::Builder::new()
         .name("EthercatTxRxThread".to_owned())
         .spawn(move || {
             #[cfg(all(target_os = "linux", not(feature = "development-build")))]
-            match set_irq_affinity(&interface, 3) {
+            match set_irq_affinity(&cloned_interface, 3) {
                 Ok(_) => tracing::info!("ethernet interrupt handler now runs on cpu:{}", 3),
                 Err(e) => tracing::error!("set_irq_handler_affinity failed: {:?}", e),
             }
-
             // Set core affinity to 4th core
             let _ = set_core_affinity(3);
-
             // Set the thread to real-time priority
             let _ = set_realtime_priority();
 
@@ -60,18 +82,18 @@ pub async fn setup_loop(interface: &str, app_state: Arc<AppState>) -> Result<(),
                 use futures::executor::block_on;
 
                 let rt = smol::LocalExecutor::new();
-                let _ = block_on(rt.run(async {
-                    tx_rx_task(&interface, tx, rx)
+                block_on(rt.run(async {
+                    tx_rx_task(&cloned_interface, tx, rx)
                         .expect("spawn TX/RX task")
                         .await
-                }));
+                }))
             }
             #[cfg(all(target_os = "linux", feature = "io-uring"))]
             {
                 use ethercrab::std::tx_rx_task_io_uring;
 
-                let _ = tx_rx_task_io_uring(&interface, tx, rx)
-                    .expect("Failed to spawn TX/RX task (io_uring)");
+                tx_rx_task_io_uring(&cloned_interface, tx, rx)
+                    .expect("Failed to spawn TX/RX task (io_uring)")
             }
         })
         .expect("Building thread");
@@ -80,64 +102,43 @@ pub async fn setup_loop(interface: &str, app_state: Arc<AppState>) -> Result<(),
     let maindevice = MainDevice::new(
         pdu,
         Timeouts {
-            // Default 5000ms
             state_transition: Duration::from_millis(5000),
-            // Default 30_000us
             pdu: Duration::from_micros(30_000),
-            // Default 10ms
-            eeprom: Duration::from_millis(100),
-            // Default 0ms
+            eeprom: Duration::from_millis(10),
             wait_loop_delay: Duration::from_millis(0),
-            // Default 100ms
-            mailbox_echo: Duration::from_millis(1000),
-            // Default 1000ms
-            mailbox_response: Duration::from_millis(10000),
+            mailbox_echo: Duration::from_millis(100),
+            mailbox_response: Duration::from_millis(1000),
         },
         MainDeviceConfig {
-            // Default RetryBehaviour::None
             retry_behaviour: RetryBehaviour::Count(5),
-            // Default 10_000
             dc_static_sync_iterations: 10_000,
         },
     );
 
-    smol::block_on({
-        let app_state_clone = app_state.clone();
-        async move {
-            let main_namespace = &mut app_state_clone
-                .socketio_setup
-                .namespaces
-                .write()
-                .await
-                .main_namespace;
-            let event = EthercatDevicesEventBuilder().initializing();
-            main_namespace.emit(MainNamespaceEvents::EthercatDevicesEvent(event));
-        }
-    });
+    // Emit "initializing" event
+    {
+        let mut main_namespace = &mut app_state
+            .socketio_setup
+            .namespaces
+            .write()
+            .await
+            .main_namespace;
+        let event = EthercatDevicesEventBuilder().initializing();
+        main_namespace.emit(MainNamespaceEvents::EthercatDevicesEvent(event));
+    }
 
-    // Initalize subdevices
-    // Fails if DC setup detects a mispatching working copunter, then just try again in loop
-    let mut group_preop = match maindevice
+    // Initialize subdevices
+    let mut group_preop = maindevice
         .init_single_group::<MAX_SUBDEVICES, PDI_LEN>(ethercat_now)
         .await
-    {
-        Ok(group) => {
-            tracing::info!("Initialized {} subdevices", &group.len());
-            group
-        }
-        Err(err) => Err(anyhow::anyhow!(
-            "[{}::setup_loop] Failed to initialize subdevices: {:?}",
-            module_path!(),
-            err
-        ))?,
-    };
+        .map_err(|e| anyhow::anyhow!("Failed to initialize subdevices: {:?}", e))?;
 
-    // create devices
+    // Create devices
     let devices =
         devices_from_subdevices::<MAX_SUBDEVICES, PDI_LEN>(&mut group_preop, &maindevice)?;
     let subdevices = group_preop.iter(&maindevice).collect::<Vec<_>>();
 
-    // extract device identifications
+    // Extract device identifications
     let device_identifications = read_device_identifications(&subdevices, &maindevice)
         .await
         .into_iter()
@@ -152,46 +153,30 @@ pub async fn setup_loop(interface: &str, app_state: Arc<AppState>) -> Result<(),
             },
         )
         .collect::<Vec<_>>();
-    let devices = device_identifications
+
+    let devices_tuple = device_identifications
         .into_iter()
         .zip(devices)
         .zip(&subdevices)
         .map(|((a, b), c)| (a, b, c))
         .collect::<Vec<_>>();
-    // filter devices and if Option<DeviceMachineIdentification> is Some
-    // return identified_devices, identified_device_identifications, identified_subdevices
+
     let (identified_device_identifications, identified_devices, identified_subdevices): (
         Vec<_>,
         Vec<_>,
         Vec<_>,
-    ) = devices
+    ) = devices_tuple
         .iter()
-        .filter(|(device_identification, _, _)| {
-            match device_identification {
-                DeviceIdentification {
-                    device_machine_identification: Some(_),
-                    ..
-                } => true,
-                _ => false
-            }
-        })
-        .map(|(device_identification, device, subdevice)| {
-            (
-                device_identification.clone(),
-                device.clone(),
-                *subdevice,
-            )
-        })
-        .fold(
-            (Vec::new(), Vec::new(), Vec::new()),
-            |mut acc, (identified_device_identification, identified_device, identified_subdevice)| {
-                acc.0.push(identified_device_identification);
-                acc.1.push(identified_device);
-                acc.2.push(identified_subdevice);
-                acc
-            },
-        );
-    // construct machines
+        .filter(|(id, _, _)| id.device_machine_identification.is_some())
+        .map(|(id, dev, sub)| (id.clone(), dev.clone(), *sub))
+        .fold((Vec::new(), Vec::new(), Vec::new()), |mut acc, x| {
+            acc.0.push(x.0);
+            acc.1.push(x.1);
+            acc.2.push(x.2);
+            acc
+        });
+
+    // Construct machines
     {
         let mut machines_guard = app_state.machines.write().await;
         machines_guard.set_ethercat_devices::<MAX_SUBDEVICES, PDI_LEN>(
@@ -206,68 +191,61 @@ pub async fn setup_loop(interface: &str, app_state: Arc<AppState>) -> Result<(),
         );
     }
 
-    // remove subdevice from devices tuple
-    let devices = devices
+    // Remove subdevice from devices tuple for app_state
+    let devices_clean = devices_tuple
         .iter()
-        .map(|(device_identification, device, _)| (device_identification.clone(), device.clone()))
+        .map(|(id, dev, _)| (id.clone(), dev.clone()))
         .collect::<Vec<_>>();
 
-    for subdevice in subdevices.iter() {
-        if subdevice.name() == "EL5152" {
-            subdevice.sdo_write(SM_OUTPUT, 0x1, 0x00u16).await?; //set sync mode (1) for free run (0)
-            subdevice.sdo_write(SM_INPUT, 0x1, 0x00u16).await?; //set sync mode (1) for free run (0)
-        }
-    }
-    // Notify client via socketio
-    let app_state_clone = app_state.clone();
-    smol::block_on(async move {
-        let main_namespace = &mut app_state_clone
+    // Emit Machines event
+    {
+        let mut main_namespace = &mut app_state
             .socketio_setup
             .namespaces
             .write()
             .await
             .main_namespace;
-        let event = MachinesEventBuilder().build(app_state_clone.clone());
+        let event = MachinesEventBuilder().build(app_state.clone());
         main_namespace.emit(MainNamespaceEvents::MachinesEvent(event));
-    });
+    }
 
     // Put group in operational state
-    let group_op = match group_preop.into_op(&maindevice).await {
-        Ok(group_op) => {
-            tracing::info!("Group in OP state");
-            group_op
-        }
-        Err(err) => Err(anyhow::anyhow!(
-            "[{}::setup_loop] Failed to put group in OP state: {:?}",
-            module_path!(),
-            err
-        ))?,
-    };
+    let group_op = group_preop
+        .into_op(&maindevice)
+        .await
+        .map_err(|e| anyhow::anyhow!("Failed to put group in OP state: {:?}", e))?;
 
-    // Write all this stuff to `app_state`
+    // Save final setup into app_state
+    *app_state.ethercat_setup.write().await = Some(EthercatSetup {
+        devices: devices_clean,
+        group: group_op,
+        maindevice,
+    });
+
+    // Emit final EthercatDevicesEvent
     {
-        let mut ethercat_setup_guard = app_state.ethercat_setup.write().await;
-        *ethercat_setup_guard = Some(EthercatSetup {
-            devices,
-            group: group_op,
-            maindevice,
-        });
-    }
-
-    // Notify client via socketio
-    let app_state_clone = app_state.clone();
-    smol::block_on(async move {
-        let main_namespace = &mut app_state_clone
+        let mut main_namespace = &mut app_state
             .socketio_setup
             .namespaces
             .write()
             .await
             .main_namespace;
-        let event = EthercatDevicesEventBuilder()
-            .build(app_state_clone.clone())
-            .await;
+        let event = EthercatDevicesEventBuilder().build(app_state.clone()).await;
         main_namespace.emit(MainNamespaceEvents::EthercatDevicesEvent(event));
-    });
+    }
+    let res: Result<(PduTx<'_>, PduRx<'_>), ethercrab::error::Error> =
+        ethercat_thread.join().expect("Thread panicked");
 
-    Ok(())
+    let tx_rx_tuple = match res {
+        Ok(tuple) => tuple,
+        Err(e) => return Err(anyhow::anyhow!("HELP")),
+    };
+
+    let ethercat_backend = EtherCatBackend {
+        tx: tx_rx_tuple.0,
+        rx: tx_rx_tuple.1,
+        interface,
+    };
+
+    Ok(ethercat_backend)
 }

server/src/loop.rs

@@ -1,9 +1,10 @@
 use crate::app_state::AppState;
+use crate::ethercat::setup::{EtherCatBackend, setup_loop};
 use bitvec::prelude::*;
 use control_core::machines::connection::MachineConnection;
 use control_core::realtime::{set_core_affinity, set_realtime_priority};
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 use tracing::{instrument, trace_span};
 
 pub fn init_loop(app_state: Arc<AppState>) -> Result<(), anyhow::Error> {
@@ -11,12 +12,20 @@ pub fn init_loop(app_state: Arc<AppState>) -> Result<(), anyhow::Error> {
     std::thread::Builder::new()
         .name("loop".to_owned())
         .spawn(move || {
+            // Before we do anything else setup the Ethercat Network
+            let res: Result<EtherCatBackend, anyhow::Error> =
+                smol::block_on(async { setup_loop("eno1", app_state.clone()).await });
+
+            let ethercat_backend: EtherCatBackend = match res {
+                Ok(ethercat_backend) => ethercat_backend,
+                Err(e) => return e,
+            };
+            let mut tx = ethercat_backend.tx;
+            let mut rx = ethercat_backend.rx;
+
             let rt = smol::LocalExecutor::new();
-
-            // Set core affinity to third core
-            let _ = set_core_affinity(2);
-
-            // Set the thread to real-time priority
+            // Set core affinity to fourth core
+            let _ = set_core_affinity(3);
             if let Err(e) = set_realtime_priority() {
                 tracing::error!(
                     "[{}::init_loop] Failed to set real-time priority \n{:?}",
@@ -30,9 +39,27 @@ pub fn init_loop(app_state: Arc<AppState>) -> Result<(), anyhow::Error> {
                 );
             }
 
-            loop {
-                let res = smol::block_on(rt.run(async { loop_once(app_state.clone()).await }));
+            let result = ethercrab::std::io_uring_one_cycle::setup_tx_rx_task("eno1");
 
+            let mut tx_rx_conf = match result {
+                Ok(tx_rx_conf) => tx_rx_conf,
+                Err(_) => todo!(),
+            };
+
+            loop {
+                let tuple = match ethercrab::std::io_uring_one_cycle::tx_rx_task_io_uring_cycle(
+                    &mut tx_rx_conf,
+                    tx,
+                    rx,
+                ) {
+                    Ok((tx, rx)) => (tx, rx),
+                    Err(_) => return anyhow::anyhow!("tx_rx_task_io_uring failed"),
+                };
+
+                tx = tuple.0;
+                rx = tuple.1;
+
+                let res = smol::block_on(rt.run(async { loop_once(app_state.clone()).await }));
                 if let Err(err) = res {
                     tracing::error!("Loop failed\n{:?}", err);
                     break;
@@ -54,34 +81,16 @@ pub fn init_loop(app_state: Arc<AppState>) -> Result<(), anyhow::Error> {
 
 #[instrument(skip(app_state))]
 pub async fn loop_once<'maindevice>(app_state: Arc<AppState>) -> Result<(), anyhow::Error> {
-    // Record cycle start for performance metrics
-    {
-        let mut metrics = app_state.performance_metrics.write().await;
-        metrics.cycle_start();
-    }
-
     let ethercat_setup_guard = app_state.ethercat_setup.read().await;
 
-    // only if we have an ethercat setup
-    // - tx/rx cycle
-    // - copy inputs to devices
     if let Some(ethercat_setup) = ethercat_setup_guard.as_ref() {
         let span = trace_span!("loop_once_inputs");
         let _enter = span.enter();
 
-        // Measure tx_rx performance
-        let txrx_start = Instant::now();
         ethercat_setup
             .group
             .tx_rx(&ethercat_setup.maindevice)
             .await?;
-        let txrx_duration = txrx_start.elapsed();
-
-        // Record tx_rx performance metrics
-        {
-            let mut metrics = app_state.performance_metrics.write().await;
-            metrics.record_txrx_time(txrx_duration);
-        }
 
         // copy inputs to devices
         for (i, subdevice) in ethercat_setup

server/src/main.rs

@@ -26,13 +26,11 @@ pub mod performance_metrics;
 pub mod rest;
 pub mod serial;
 pub mod socketio;
-
 use crate::ethercat::init::init_ethercat;
 
 fn main() {
     logging::init_tracing();
     tracing::info!("Tracing initialized successfully");
-
     init_panic_handling();
 
     #[cfg(feature = "memory-locking")]
@@ -48,19 +46,14 @@ fn main() {
         .name("init".to_string())
         .spawn({
             move || {
-                #[cfg(feature = "dhat-heap")]
-                init_dhat_heap_profiling();
-
                 init_socketio_queue(app_state.clone());
                 init_api(app_state.clone()).expect("Failed to initialize API");
                 init_loop(app_state.clone()).expect("Failed to initialize loop");
 
                 #[cfg(feature = "mock-machine")]
                 init_mock(app_state.clone()).expect("Failed to initialize mock machines");
-
                 #[cfg(not(feature = "mock-machine"))]
                 init_serial(app_state.clone()).expect("Failed to initialize Serial");
-
                 #[cfg(not(feature = "mock-machine"))]
                 init_ethercat(app_state).expect("Failed to initialize EtherCAT");
             }
@@ -77,30 +70,3 @@ fn main() {
         std::thread::park();
     }
 }
-
-#[cfg(feature = "dhat-heap")]
-fn init_dhat_heap_profiling() {
-    use std::{process::exit, time::Duration};
-
-    let profiler = dhat::Profiler::new_heap();
-
-    smol::block_on(async {
-        let dhat_analysis_time = std::time::Duration::from_secs(60);
-        let dhat_analysis_start = std::time::Instant::now();
-        tracing::info!(
-            "Starting dhat heap profiler for {} seconds",
-            dhat_analysis_time.as_secs()
-        );
-
-        loop {
-            use std::time::Duration;
-            smol::Timer::after(Duration::from_secs(1)).await;
-
-            // if `dhat-heap` is enabled, we will analyze the heap for 60 seconds and then exit
-            if dhat_analysis_start.elapsed() > dhat_analysis_time {
-                drop(profiler);
-                exit(0)
-            }
-        }
-    })
-}