Async pipelined FPWR + 4kHz: exec 245us->24us, RTT 2ms->1.25-1.5ms

- FPWR fire-and-forget: two alternating leaked tx slots, or()-driven
  20us kick to reach sendable, poll_once reap after next spin-detect
  (late=0 over 350k cycles). Fixes: reap placement, poll_once kick
- exec drops to 24us mean, freeing budget for 4kHz (250us) operation
- 4kHz verified: TwinCAT PLC at 4001.8Hz, lag 5-6 cycles
  (RTT 1.25-1.5ms); J1900 effective 2.9kHz - FPRD poll cost (71us x N)
  exceeds 250us budget, occasional frame miss
- Conclusion: J1900 write wait never contributed to lag; the 4-stage
  pipeline is TwinCAT IO mapping + bridge forwarding

scripts/tc-deploy-retry-4khz.ps1

@@ -0,0 +1,61 @@
+#Requires -RunAsAdministrator
+#Requires -Version 5.1
+$ErrorActionPreference = 'Continue'
+
+function Invoke-Com {
+    param([scriptblock]$Fn, [int]$Retries = 30, [int]$DelayMs = 2000)
+    for ($i = 0; $i -lt $Retries; $i++) {
+        try { return & $Fn } catch {
+            $hr = $_.Exception.HResult
+            if ($hr -eq -2147418111) { Start-Sleep -Milliseconds $DelayMs; continue }
+            throw
+        }
+    }
+    throw "COM call rejected $Retries times (RPC_E_CALL_REJECTED)"
+}
+
+Write-Host "=== Deploy 8B with RPC retry ==="
+$dte = Invoke-Com { New-Object -ComObject TcXaeShell.DTE.17.0 }
+Invoke-Com { $dte.MainWindow.Visible = $true } | Out-Null
+try { Invoke-Com { $dte.SuppressUI = $false } | Out-Null } catch {}
+Invoke-Com { $dte.Solution.Open('C:\Users\tonycao\work\twincat3-auto-cli\TwinCAT_EL6695_Primary_4kHz\TwinCATProject1.sln') } | Out-Null
+Write-Host "solution opened"
+
+$sm = $null
+for ($i = 0; $i -lt 40; $i++) {
+    try { $sm = Invoke-Com { $dte.Solution.Projects.Item(1).Object } -Retries 3 -DelayMs 3000; if ($sm) { break } } catch { Write-Host "  waiting sysmgr..." }
+}
+if (-not $sm) { Write-Host "FATAL: no sysmgr"; exit 1 }
+Write-Host "sysmgr OK"
+Invoke-Com { $sm.SetTargetNetId('169.254.176.217.1.1') } | Out-Null
+
+Write-Host "building..."
+try { Invoke-Com { $dte.ExecuteCommand('Build.BuildSolution') } -Retries 10 | Out-Null } catch { Write-Host "build cmd: $_" }
+# wait for build done (BuildState 3 = done)
+$done = $false
+for ($i = 0; $i -lt 90; $i++) {
+    try {
+        $bs = Invoke-Com { $dte.Solution.SolutionBuild.BuildState } -Retries 5 -DelayMs 3000
+        $bi = Invoke-Com { $dte.Solution.SolutionBuild.LastBuildInfo } -Retries 5 -DelayMs 1000
+        if ($bs -eq 3) { Write-Host "build done (LastBuildInfo=$bi) after $($i*2)s"; $done = $true; break }
+    } catch { Write-Host "  build poll err: $_" }
+    Start-Sleep -Seconds 2
+}
+if (-not $done) { Write-Host "WARN: build state never reached done" }
+
+# dump build errors if any
+try {
+    $ew = $dte.ToolWindows.ErrorList
+    Write-Host "error list check skipped"
+} catch {}
+
+Write-Host "activating..."
+Invoke-Com { $sm.ActivateConfiguration() } -Retries 30 -DelayMs 3000 | Out-Null
+Write-Host "activated"
+Start-Sleep 10
+Write-Host "restarting runtime..."
+Invoke-Com { $sm.StartRestartTwinCAT() } -Retries 30 -DelayMs 3000 | Out-Null
+Write-Host "restart issued"
+Start-Sleep 12
+try { Write-Host "IsTwinCATStarted=$($sm.IsTwinCATStarted)" } catch {}
+Write-Host "=== done ==="

src/main.rs

@@ -755,6 +755,17 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
     let mut prev_rx8 = [0u8; 8];       // spin-lock: previous payload (change detect)
     let mut last_fresh: Option<Instant> = None; // spin-lock: last fresh-data instant
 
+    // Async pipelined FPWR: two alternating leaked tx slots so the in-flight
+    // write future can hold a 'static buffer while the response is reaped
+    // at the start of the next cycle.
+    let tx_slot_a = Box::leak(Box::new([0u8; 8]));
+    let tx_slot_b = Box::leak(Box::new([0u8; 8]));
+    let mut tx_slot_toggle = false;
+    let mut pending_tx: Option<
+        std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), ethercrab::error::Error>> + '_>>,
+    > = None;
+    let mut tx_late: u64 = 0;
+
     let mut aborted: Option<String> = None;
 
     loop {
@@ -823,6 +834,21 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
         let t0 = Instant::now();
         cyc += 1;
 
+        // Reap async FPWR kicked last cycle: the spin-detect above gave the
+        // response ~one full cycle to arrive, so this should never block.
+        if let Some(mut f) = pending_tx.take() {
+            match futures_lite::future::poll_once(f.as_mut()).await {
+                Some(Ok(_)) => {}
+                Some(Err(_)) => {
+                    tx_err += 1;
+                }
+                None => {
+                    tx_late += 1;
+                    let _ = f.await;
+                }
+            }
+        }
+
         if !spin_mode {
             // 1. Read SM3 (TwinCAT data via bridge)
             let t_rx0 = Instant::now();
@@ -913,14 +939,47 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
                 tx_buf[264..268].copy_from_slice(&(jitter_ns as i32).to_le_bytes());
             }
 
-            let t_tx0 = Instant::now();
-            let tx_ok = ethercrab::Command::fpwr(station, SM_TX_ADDR)
-                .send(&maindevice, &tx_buf[..PAYLOAD_LEN])
-                .await
-                .is_ok();
-            tx_rtt_st.push(t_tx0.elapsed().as_nanos() as i64);
-            if !tx_ok {
-                tx_err += 1;
+            if spin_mode && PAYLOAD_LEN == 8 {
+                // Async pipelined FPWR: kick the frame onto the wire now
+                // (poll_once drives it to sendable), reap response next
+                // cycle. Frees ~240us of blocking write time per cycle.
+                let p = if tx_slot_toggle {
+                    tx_slot_b.as_mut_ptr()
+                } else {
+                    tx_slot_a.as_mut_ptr()
+                };
+                tx_slot_toggle = !tx_slot_toggle;
+                let slot: &'static mut [u8] =
+                    unsafe { std::slice::from_raw_parts_mut(p, PAYLOAD_LEN) };
+                slot.copy_from_slice(&tx_buf[..PAYLOAD_LEN]);
+                let mut fut = Box::pin(
+                    ethercrab::Command::fpwr(station, SM_TX_ADDR).send(&maindevice, &slot[..]),
+                );
+                // Kick: drive the write future for up to ~20us so the frame
+                // is marked sendable and the TX thread puts it on the wire
+                // NOW. poll_once alone does not reach sendable state.
+                let t_kick = Instant::now();
+                let _ = futures_lite::future::or(
+                    async { fut.as_mut().await.is_ok() },
+                    async {
+                        while t_kick.elapsed().as_micros() < 20 {
+                            futures_lite::future::yield_now().await;
+                        }
+                        false
+                    },
+                )
+                .await;
+                pending_tx = Some(fut);
+            } else {
+                let t_tx0 = Instant::now();
+                let tx_ok = ethercrab::Command::fpwr(station, SM_TX_ADDR)
+                    .send(&maindevice, &tx_buf[..PAYLOAD_LEN])
+                    .await
+                    .is_ok();
+                tx_rtt_st.push(t_tx0.elapsed().as_nanos() as i64);
+                if !tx_ok {
+                    tx_err += 1;
+                }
             }
         }
 
@@ -1051,13 +1110,14 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
             let proc_val = tc_seq.wrapping_mul(7).wrapping_add(3);
             let stale_pct = if cyc > 0 { (rx_stale * 100 / cyc) } else { 0 };
             println!(
-                "S tms={} cyc={} raw={} proc={} txe={} rxe={} stale={}% mono={} cerr={} mism={} e0={} e1={}",
+                "S tms={} cyc={} raw={} proc={} txe={} rxe={} late={} stale={}% mono={} cerr={} mism={} e0={} e1={}",
                 start.elapsed().as_millis(),
                 cyc,
                 tc_seq,
                 proc_val,
                 tx_err,
                 rx_err,
+                tx_late,
                 stale_pct,
                 rx_nonmono,
                 comp_err,
@@ -1073,6 +1133,11 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
         }
     }
 
+    // Reap any in-flight async FPWR before reporting
+    if let Some(f) = pending_tx.take() {
+        let _ = f.await;
+    }
+
     // ---- Final report ----
     println!("\n=== timing stats ===");
     jitter_st.report(1000, "us");