From ac99194b4e8e68b7443d9c0150a70b5782b20197 Mon Sep 17 00:00:00 2001
From: Tony Cao <tonycao@kanghexin.work>
Date: Mon, 26 Oct 2026 18:57:06 +0800
Subject: [PATCH] wip(pipeline): async non-blocking FPWR fails - executor
 starvation identified

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).
---
 src/main.rs | 34 ++++++++++++++++++++++++----------
 1 file changed, 24 insertions(+), 10 deletions(-)

diff --git a/src/main.rs b/src/main.rs
index 0c6478c..1f24c22 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -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_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 j2cx_pending: Option<std::time::Instant> = None;
 
     while !STOP.load(Ordering::Relaxed) {
         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(); }
                     wphase_st.push(((rt::now_ns() % 1_000_000) / 1000) as i64);
                 }
-                let tx_ok = ethercrab::Command::fpwr(station, SM_TX_ADDR)
-                    .send(maindevice, &slot[..PAYLOAD_LEN])
-                    .await
-                    .is_ok();
-                n_fpwr += 1;
-                if tx_ok {
-                    j2cx_st.push(t_detect.elapsed().as_nanos() as i64);
-                } else {
-                    n_fpwr_err += 1;
-                }
+                // PIPELINE v2: mirror the fprd's working pattern - a short
+                // yield-kick lets the executor run the TX task (frame goes on
+                // the wire), then store for poll-based completion at reap.
+                let mut f: FpwrFut<'_> = Box::pin(
+                    ethercrab::Command::fpwr(station, SM_TX_ADDR).send(maindevice, &slot[..PAYLOAD_LEN]),
+                );
+                let t_kick = Instant::now();
+                let _ = futures_lite::future::or(
+                    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 {
                 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() {
                     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; }