tonycao/ethercat-linux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

wip(pipeline): async non-blocking FPWR fails - executor starvation identified

Browse Source 
Tried in async event loop: (a) issue-and-poll-once, (b) 20us yield-kick
mirror of working fprd pattern - both complete futures with ferr=0 yet
frames never reach the terminal (box input zero mid-run). Root cause:
poll_once-based loop never truly yields to the executor, TX/RX tasks
starve, frame lifecycle stalls. Bridge mode works because its fprd uses
blocking await (real yields). Full fix = tokio::select!-based concurrent
task rewrite (fprd + fpwr + hop) - future work package.

NOTE: async mode on this branch is BROKEN (use bridge mode, or d2000
branch build e57665be for working async).
This commit is contained in:
Tony Cao 2026-10-26 18:57:06 +08:00
parent 788499cf55
commit ac99194b4e
1 changed files with 24 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
src/main.rs
View File

@ -2387,6 +2387,7 @@ async fn rfm_async_loop(
let mut phase_st = OnlineStats::new("fresh_phase_us", 200, -1_500_000, 1_500_000); let mut phase_st = OnlineStats::new("fresh_phase_us", 200, -1_500_000, 1_500_000);
let mut phase_est: i64 = 0; // estimated TwinCAT frame boundary phase (ns within 1ms) let mut phase_est: i64 = 0; // estimated TwinCAT frame boundary phase (ns within 1ms)
let mut wphase_st = OnlineStats::new("write_phase_us", 200, -1_500_000, 1_500_000); let mut wphase_st = OnlineStats::new("write_phase_us", 200, -1_500_000, 1_500_000);
let mut j2cx_pending: Option<std::time::Instant> = None;
while !STOP.load(Ordering::Relaxed) { while !STOP.load(Ordering::Relaxed) {
if duration_sec > 0 && start.elapsed().as_secs() >= duration_sec { if duration_sec > 0 && start.elapsed().as_secs() >= duration_sec {
@ -2549,16 +2550,26 @@ async fn rfm_async_loop(
while (rt::now_ns() as i64) < target { core::hint::spin_loop(); } while (rt::now_ns() as i64) < target { core::hint::spin_loop(); }
wphase_st.push(((rt::now_ns() % 1_000_000) / 1000) as i64); wphase_st.push(((rt::now_ns() % 1_000_000) / 1000) as i64);
} }
let tx_ok = ethercrab::Command::fpwr(station, SM_TX_ADDR) // PIPELINE v2: mirror the fprd's working pattern - a short
.send(maindevice, &slot[..PAYLOAD_LEN]) // yield-kick lets the executor run the TX task (frame goes on
.await // the wire), then store for poll-based completion at reap.
.is_ok(); let mut f: FpwrFut<'_> = Box::pin(
n_fpwr += 1; ethercrab::Command::fpwr(station, SM_TX_ADDR).send(maindevice, &slot[..PAYLOAD_LEN]),
if tx_ok { );
j2cx_st.push(t_detect.elapsed().as_nanos() as i64); let t_kick = Instant::now();
} else { let _ = futures_lite::future::or(
n_fpwr_err += 1; async { f.as_mut().await.is_ok() },
} async {
while t_kick.elapsed().as_micros() < 20 {
futures_lite::future::yield_now().await;
}
false
},
)
.await;
pending_fpwr = Some(f);
j2cx_pending = Some(t_detect);
n_fpwr += 1; // issued; result counted at reap
} else { } else {
n_ovf += 1; // wire busy; newer tx_latest will go out after reap n_ovf += 1; // wire busy; newer tx_latest will go out after reap
} }
@ -2635,6 +2646,9 @@ async fn rfm_async_loop(
if res.is_err() { if res.is_err() {
n_fpwr_err += 1; n_fpwr_err += 1;
} }
if let Some(td) = j2cx_pending.take() {
j2cx_st.push(td.elapsed().as_nanos() as i64);
}
} }
} }
{ let n = Instant::now(); seg_fpwr_us = n.duration_since(cursor).as_micros() as u64; cursor = n; } { let n = Instant::now(); seg_fpwr_us = n.duration_since(cursor).as_micros() as u64; cursor = n; }