rewa81/qitech-control
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #369 from qitechgmbh/367-retry-ethercat-txrx-if-it-fails

Browse Source 
Retry EtherCAT TX/RX if it fails
This commit is contained in:
Icedsoda authored and GitHub committed 2025-06-17 16:41:24 +02:00
2 parents 7c8b05241a 2ed2361fdf
commit 0cfd7eec9f
4 files changed
+422 -9
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

control-core/src/helpers/retry.rs
+415 -1
View File
@@ -1,4 +1,38 @@
pub fn retry<T, E, F>(n: usize, mut f: F) -> Result<T, E>
use std::future::Future;
/// Retries a fallible operation up to `n` times synchronously.
///
/// This function executes the provided closure `f` repeatedly until either:
/// - The operation succeeds (returns `Ok(T)`)
/// - All retry attempts are exhausted
///
/// # Arguments
///
/// * `n` - The maximum number of retry attempts (0 means only one attempt)
/// * `f` - A mutable closure that returns a `Result<T, E>`
///
/// # Returns
///
/// Returns `Ok(T)` if any attempt succeeds, or `Err(E)` containing the error
/// from the last failed attempt if all retries are exhausted.
///
/// # Examples
///
/// ```rust
/// use control_core::helpers::retry::retry_n_times;
///
/// let mut attempt = 0;
/// let result = retry_n_times(3, || {
/// attempt += 1;
/// if attempt < 3 {
/// Err("failed")
/// } else {
/// Ok("success")
/// }
/// });
/// assert_eq!(result, Ok("success"));
/// ```
pub fn retry_n_times<T, E, F>(n: usize, mut f: F) -> Result<T, E>
where
F: FnMut() -> Result<T, E>,
{
@@ -11,3 +45,383 @@ where
}
Err(last_err.unwrap())
}
/// Retries a fallible async operation up to `n` times.
///
/// This function executes the provided async closure `f` repeatedly until either:
/// - The operation succeeds (returns `Ok(T)`)
/// - All retry attempts are exhausted
///
/// Each attempt is awaited before proceeding to the next retry or returning.
///
/// # Arguments
///
/// * `n` - The maximum number of retry attempts (0 means only one attempt)
/// * `f` - A mutable closure that returns a `Future<Output = Result<T, E>>`
///
/// # Returns
///
/// Returns `Ok(T)` if any attempt succeeds, or `Err(E)` containing the error
/// from the last failed attempt if all retries are exhausted.
///
/// # Examples
///
/// ```rust
/// use control_core::helpers::retry::retry_n_times_async;
/// use std::sync::Arc;
/// use std::sync::atomic::{AtomicUsize, Ordering};
///
/// async fn example() {
/// let attempt = Arc::new(AtomicUsize::new(0));
/// let attempt_clone = attempt.clone();
/// let result = retry_n_times_async(3, move || {
/// let attempt = attempt_clone.clone();
/// async move {
/// let current = attempt.fetch_add(1, Ordering::SeqCst) + 1;
/// if current < 3 {
/// Err("failed")
/// } else {
/// Ok("success")
/// }
/// }
/// }).await;
/// assert_eq!(result, Ok("success"));
/// }
/// ```
pub async fn retry_n_times_async<T, E, F, Fut>(n: usize, mut f: F) -> Result<T, E>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<T, E>>,
{
let mut last_err: Option<E> = None;
for _ in 0..=n {
match f().await {
Ok(value) => return Ok(value),
Err(err) => last_err = Some(err),
}
}
Err(last_err.unwrap())
}
/// Retries a fallible operation until a closure decides to stop retrying.
///
/// This function executes the provided closure `f` repeatedly until either:
/// - The operation succeeds (returns `Ok(T)`)
/// - The retry condition returns false
///
/// # Arguments
///
/// * `f` - A mutable closure that returns a `Result<T, E>`
/// * `should_retry` - A closure that takes `&E` and returns `bool` to decide if retry should continue
///
/// # Returns
///
/// Returns `Ok(T)` if any attempt succeeds, or `Err(E)` containing the error
/// from the last failed attempt when retries stop.
///
/// # Examples
///
/// ```rust
/// use control_core::helpers::retry::retry_conditionally;
///
/// let mut attempt = 0;
/// let result = retry_conditionally(
/// || {
/// attempt += 1;
/// if attempt < 3 {
/// Err(format!("failed attempt {}", attempt))
/// } else {
/// Ok("success")
/// }
/// },
/// |err| {
/// // Retry up to 3 times
/// !err.contains("attempt 3")
/// }
/// );
/// assert_eq!(result, Ok("success"));
/// ```
pub fn retry_conditionally<T, E, F, R>(mut f: F, mut should_retry: R) -> Result<T, E>
where
F: FnMut() -> Result<T, E>,
R: FnMut(&E) -> bool,
{
loop {
match f() {
Ok(value) => return Ok(value),
Err(err) => {
if !should_retry(&err) {
return Err(err);
}
}
}
}
}
/// Retries a fallible async operation until a closure decides to stop retrying.
///
/// This function executes the provided async closure `f` repeatedly until either:
/// - The operation succeeds (returns `Ok(T)`)
/// - The retry condition returns false
///
/// # Arguments
///
/// * `f` - A mutable closure that returns a `Future<Output = Result<T, E>>`
/// * `should_retry` - A closure that takes `&E` and returns `bool` to decide if retry should continue
///
/// # Returns
///
/// Returns `Ok(T)` if any attempt succeeds, or `Err(E)` containing the error
/// from the last failed attempt when retries stop.
///
/// # Examples
///
/// ```rust
/// use control_core::helpers::retry::retry_conditionally_async;
/// use std::sync::Arc;
/// use std::sync::atomic::{AtomicUsize, Ordering};
///
/// async fn example() {
/// let attempt = Arc::new(AtomicUsize::new(0));
/// let attempt_clone = attempt.clone();
/// let result = retry_conditionally_async(
/// move || {
/// let attempt = attempt_clone.clone();
/// async move {
/// let current = attempt.fetch_add(1, Ordering::SeqCst) + 1;
/// if current < 3 {
/// Err(format!("failed attempt {}", current))
/// } else {
/// Ok("success")
/// }
/// }
/// },
/// |err| {
/// // Retry up to 3 times
/// !err.contains("attempt 3")
/// }
/// ).await;
/// assert_eq!(result, Ok("success"));
/// }
/// ```
pub async fn retry_conditionally_async<T, E, F, Fut, R>(
mut f: F,
mut should_retry: R,
) -> Result<T, E>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<T, E>>,
R: FnMut(&E) -> bool,
{
loop {
match f().await {
Ok(value) => return Ok(value),
Err(err) => {
if !should_retry(&err) {
return Err(err);
}
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::cell::RefCell;
use std::rc::Rc;
// Simple blocking executor for async functions in tests
fn block_on<F: std::future::Future>(future: F) -> F::Output {
use std::sync::Arc;
use std::task::{Context, Poll, Waker};
use std::thread;
struct DummyWaker;
impl std::task::Wake for DummyWaker {
fn wake(self: Arc<Self>) {}
}
let waker = Waker::from(Arc::new(DummyWaker));
let mut context = Context::from_waker(&waker);
let mut future = Box::pin(future);
loop {
match future.as_mut().poll(&mut context) {
Poll::Ready(result) => return result,
Poll::Pending => {
thread::yield_now();
}
}
}
}
#[test]
fn test_retry_success_first_attempt() {
let result = retry_n_times(3, || Ok::<i32, &str>(42));
assert_eq!(result, Ok(42));
}
#[test]
fn test_retry_success_after_failures() {
let mut attempt = 0;
let result = retry_n_times(3, || {
attempt += 1;
if attempt < 3 { Err("failed") } else { Ok(42) }
});
assert_eq!(result, Ok(42));
assert_eq!(attempt, 3);
}
#[test]
fn test_retry_all_attempts_fail() {
let mut attempt = 0;
let result: Result<i32, &str> = retry_n_times(2, || {
attempt += 1;
Err("always fails")
});
assert_eq!(result, Err("always fails"));
assert_eq!(attempt, 3); // 0..=2 means 3 attempts
}
#[test]
fn test_retry_zero_retries() {
let mut attempt = 0;
let result: Result<i32, &str> = retry_n_times(0, || {
attempt += 1;
Err("failed")
});
assert_eq!(result, Err("failed"));
assert_eq!(attempt, 1); // Only one attempt
}
#[test]
fn test_retry_async_success_first_attempt() {
let result = block_on(retry_n_times_async(3, || async { Ok::<i32, &str>(42) }));
assert_eq!(result, Ok(42));
}
#[test]
fn test_retry_async_success_after_failures() {
let attempt = Rc::new(RefCell::new(0));
let attempt_clone = attempt.clone();
let result = block_on(retry_n_times_async(3, move || {
let attempt = attempt_clone.clone();
async move {
let mut a = attempt.borrow_mut();
*a += 1;
if *a < 3 { Err("failed") } else { Ok(42) }
}
}));
assert_eq!(result, Ok(42));
assert_eq!(*attempt.borrow(), 3);
}
#[test]
fn test_retry_async_all_attempts_fail() {
let attempt = Rc::new(RefCell::new(0));
let attempt_clone = attempt.clone();
let result: Result<i32, &str> = block_on(retry_n_times_async(2, move || {
let attempt = attempt_clone.clone();
async move {
let mut a = attempt.borrow_mut();
*a += 1;
Err("always fails")
}
}));
assert_eq!(result, Err("always fails"));
assert_eq!(*attempt.borrow(), 3); // 0..=2 means 3 attempts
}
#[test]
fn test_retry_and_success_with_condition() {
let mut attempt = 0;
let result: Result<i32, String> = retry_conditionally(
|| {
attempt += 1;
if attempt < 3 {
Err(format!("failed attempt {}", attempt))
} else {
Ok(42)
}
},
|err| {
// Retry as long as it's not the 3rd attempt
!err.contains("attempt 3")
},
);
assert_eq!(result, Ok(42));
assert_eq!(attempt, 3);
}
#[test]
fn test_retry_and_stop_early() {
let mut attempt = 0;
let result: Result<i32, String> = retry_conditionally(
|| {
attempt += 1;
Err(format!("failed attempt {}", attempt))
},
|_err| {
// Stop after first failure
false
},
);
assert!(result.is_err());
assert_eq!(attempt, 1);
assert_eq!(result.unwrap_err(), "failed attempt 1");
}
#[test]
fn test_retry_async_success_with_condition() {
let attempt = Rc::new(RefCell::new(0));
let attempt_clone = attempt.clone();
let result: Result<i32, String> = block_on(retry_conditionally_async(
move || {
let attempt = attempt_clone.clone();
async move {
let mut a = attempt.borrow_mut();
*a += 1;
if *a < 3 {
Err(format!("failed attempt {}", *a))
} else {
Ok(42)
}
}
},
|err| {
// Retry as long as it's not the 3rd attempt
!err.contains("attempt 3")
},
));
assert_eq!(result, Ok(42));
assert_eq!(*attempt.borrow(), 3);
}
#[test]
fn test_retry_async_stop_early() {
let attempt = Rc::new(RefCell::new(0));
let attempt_clone = attempt.clone();
let result: Result<i32, String> = block_on(retry_conditionally_async(
move || {
let attempt = attempt_clone.clone();
async move {
let mut a = attempt.borrow_mut();
*a += 1;
Err(format!("failed attempt {}", *a))
}
},
|_err| {
// Stop after first failure
false
},
));
assert!(result.is_err());
assert_eq!(*attempt.borrow(), 1);
assert_eq!(result.unwrap_err(), "failed attempt 1");
}
}
server/src/ethercat/setup.rs
+2 -2
View File
@@ -59,7 +59,7 @@ pub async fn setup_loop(
// Default 5000ms
state_transition: Duration::from_millis(10 * 1000),
// Default 30_000us
pdu: Duration::from_millis(1000),
pdu: Duration::from_millis(25),
// Default 10ms
eeprom: Duration::from_millis(10),
// Default 0ms
@@ -71,7 +71,7 @@ pub async fn setup_loop(
},
MainDeviceConfig {
// Default RetryBehaviour::None
retry_behaviour: RetryBehaviour::None,
retry_behaviour: RetryBehaviour::Count(100), // 25ms * 100 = 2.5s
// Default 10_000
dc_static_sync_iterations: 10_000,
},
server/src/loop.rs
+3 -4
View File
@@ -2,11 +2,11 @@ use crate::app_state::AppState;
use crate::panic::{PanicDetails, send_panic};
use bitvec::prelude::*;
use control_core::helpers::loop_trottle::LoopThrottle;
use control_core::helpers::retry::retry_conditionally_async;
use smol::channel::Sender;
use std::sync::Arc;
use std::time::Duration;
use tracing::{info_span, instrument, trace_span};
use tracing_futures::Instrument as _;
use tracing::{debug, error, instrument, trace_span};
pub fn init_loop(
thread_panic_tx: Sender<PanicDetails>,
@@ -18,7 +18,7 @@ pub fn init_loop(
.spawn(move || {
send_panic(thread_panic_tx.clone());
let rt = smol::LocalExecutor::new();
let mut throttle = LoopThrottle::new(Duration::from_millis(1), 1, None);
let mut throttle = LoopThrottle::new(Duration::from_millis(1), 10, None);
loop {
let res = smol::block_on(rt.run(async {
throttle.sleep().await;
@@ -59,7 +59,6 @@ pub async fn loop_once<'maindevice>(app_state: Arc<AppState>) -> Result<(), anyh
let span = trace_span!("loop_once_inputs");
let _enter = span.enter();
// TX/RX cycle
ethercat_setup
.group
.tx_rx(&ethercat_setup.maindevice)
server/src/serial/devices/laser/mod.rs
+2 -2
View File
@@ -10,7 +10,7 @@ use anyhow::anyhow;
use control_core::{
helpers::{
hashing::{hashing, xor_u128_to_u16},
retry::retry,
retry::retry_n_times,
},
machines::identification::{
DeviceHardwareIdentification, DeviceHardwareIdentificationSerial, DeviceIdentification,
@@ -174,7 +174,7 @@ impl Laser {
loop {
// send diameter request
let response = retry(10, || {
let response = retry_n_times(10, || {
if let Err(e) = port.write_all(&request_buffer) {
return Err(anyhow!("Failed to write to port: {}", e));
}