From 010fef0b5d8207390a55475fce148a58177b85f1 Mon Sep 17 00:00:00 2001 From: Tony Cao Date: Tue, 27 Oct 2026 16:28:39 +0800 Subject: [PATCH] diagnose(pipeline): wire-capture closes the case - vendored ethercrab response cross-talk Instrumented passive AF_PACKET sniffer (--passive, LE addr fix critical): - DIRECT: SM2 writes 100% accepted (wkc1=10842/wkc0=10842, request+response pairs) - PIPELINE: only 36% accepted (wkc1=3678/wkc0=16638) - SM2 3-buffer handshake rejects premature back-to-back writes - Both modes: identical wire traffic (~830 writes/s, live payloads, 828B frames) Mechanism: concurrent fprd+fpwr futures -> vendored ethercrab response routing completes the fpwr future with the fprd's wkc=1 response (bogus Ok, ferr=0) -> write slot freed early -> next write issued while ESC busy -> wkc=0 discard -> buffer desync -> bridge reads empty buffers -> primary image zero. Prior 'executor starvation' theory DISPROVEN by wire capture. Fix scope: patch vendored ethercrab PDU index matching (or serialize EC ops - no gain). --- src/main.rs | 85 +++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 83 insertions(+), 2 deletions(-) diff --git a/src/main.rs b/src/main.rs index d720b42..8f26923 100644 --- a/src/main.rs +++ b/src/main.rs @@ -208,6 +208,7 @@ fn parse_args() -> Result { "--adopt-pure" => cfg.adopt_pure = true, "--sync-ts" => cfg.sync_ts = true, "--dc-discipline" => cfg.dc_discipline = true, + "--passive" => {}, "--sm3-len" => { cfg.sm3_len_force = val("--sm3-len")?.parse().map_err(|_| "bad --sm3-len")?; }, "--dc" => cfg.dc = true, "--iouring" => cfg.iouring = true, @@ -338,6 +339,77 @@ async fn run(cfg: &Config) -> Result<(), Box> { } }); + + if std::env::args().any(|a| a == "--passive") { + use std::ffi::CString; + let ciface = CString::new(cfg.iface.as_bytes()).unwrap(); + let fd = unsafe { libc::socket(libc::AF_PACKET, libc::SOCK_RAW, (libc::ETH_P_ALL as u16).to_be() as i32) }; + assert!(fd >= 0, "raw socket failed"); + let ifidx = unsafe { libc::if_nametoindex(ciface.as_ptr()) }; + assert!(ifidx > 0); + let mut sll: libc::sockaddr_ll = unsafe { std::mem::zeroed() }; + sll.sll_family = libc::AF_PACKET as u16; + sll.sll_protocol = (libc::ETH_P_ALL as u16).to_be(); + sll.sll_ifindex = ifidx as i32; + let r = unsafe { libc::bind(fd, &sll as *const libc::sockaddr_ll as *const libc::sockaddr, std::mem::size_of::() as u32) }; + assert!(r == 0, "bind failed"); + let tv = libc::timeval { tv_sec: 0, tv_usec: 50_000 }; + unsafe { libc::setsockopt(fd, libc::SOL_SOCKET, libc::SO_RCVTIMEO, &tv as *const _ as *const libc::c_void, std::mem::size_of::() as u32) }; + let t0 = std::time::Instant::now(); + let (mut n_fpwr, mut n_fprd, mut n_apwr, mut n_aprd, mut n_other) = (0u64, 0u64, 0u64, 0u64, 0u64); + let (mut n_sm2, mut n_other_addr) = (0u64, 0u64); + let (mut n_sm2_wkc1, mut n_sm2_wkc0) = (0u64, 0u64); + let mut samples = 0; + println!("PASSIVE: listening 12s for EtherCAT frames (FPWR cmd=0x05 to 0x1C00)?"); + while t0.elapsed() < std::time::Duration::from_secs(12) { + let mut buf = [0u8; 2048]; + let n = unsafe { libc::recv(fd, buf.as_mut_ptr() as *mut libc::c_void, buf.len(), 0) }; + if n <= 0 { continue; } + let n = n as usize; + if n < 27 { continue; } + let et = u16::from_be_bytes([buf[12], buf[13]]); + if et != 0x88A4 { continue; } + let cmd = buf[16]; + match cmd { + 0x05 => { + n_fpwr += 1; + let ado = u16::from_le_bytes([buf[20], buf[21]]); + if ado == 0x1C00 { + n_sm2 += 1; + if n >= 828 { + let wkc = u16::from_le_bytes([buf[826], buf[827]]); + if wkc == 1 { n_sm2_wkc1 += 1; } else { n_sm2_wkc0 += 1; } + } + if samples < 6 && n >= 42 { + let p = &buf[26..42]; + let hex: Vec = p.iter().map(|b| format!("{:02X}", b)).collect(); + let wkc = if n >= 828 { u16::from_le_bytes([buf[826], buf[827]]) } else { 0 }; + println!(" SM2 pay={}.. wkc={}", &hex.join(" ")[..23], wkc); + samples += 1; + } + } else { + n_other_addr += 1; + if n_other_addr % 5000 == 1 { + let adp = u16::from_le_bytes([buf[18], buf[19]]); + } + } + } + 0x04 => { + n_fprd += 1; + if n_fprd % 4000 == 1 { + let adp = u16::from_le_bytes([buf[18], buf[19]]); + let ado = u16::from_le_bytes([buf[20], buf[21]]); + } + } + 0x02 => n_apwr += 1, + 0x01 => n_aprd += 1, + _ => n_other += 1, + } + } + println!("PASSIVE done: FPWR={} (toSM2={} wkc1={} wkc0={} otherAddr={}) FPRD={}", n_fpwr, n_sm2, n_sm2_wkc1, n_sm2_wkc0, n_other_addr, n_fprd); + return Ok(()); + } + println!("Scanning segment (INIT -> PREOP)..."); let group = maindevice .init_single_group::(ethercat_now) @@ -461,7 +533,7 @@ async fn run(cfg: &Config) -> Result<(), Box> { let len = hdr & 0x0FFF; let typ = hdr >> 12; let cmd = f[16]; let adp = u16::from_be_bytes([f[18], f[19]]); let ado = u16::from_be_bytes([f[20], f[21]]); - extra = format!(" || ecat len={} type={} cmd=0x{:02X} adp={:#06X} ado={:#06X} wkc={}", len, typ, cmd, adp, ado, f[13 + len as usize]); + extra = format!(" || ecat len={} type={} cmd=0x{:02X} adp={:04X} ado={:04X} wkc={}", len, typ, cmd, adp, ado, f[13 + len as usize]); } println!("{} [{}B] {}{}", tag, n, hex.join(" "), extra); }; @@ -1557,7 +1629,7 @@ async fn run(cfg: &Config) -> Result<(), Box> { // ---- rfm-async: event-driven forwarder, no tick ---- if cfg.mode == Mode::RfmAsync { let rfm = rfm_ctx.take().expect("rfm not opened"); - return rfm_async_loop(&maindevice, station, rfm, cfg.duration_sec, cfg.dc_discipline).await; + return rfm_async_loop(&maindevice, station, rfm, cfg.duration_sec, cfg.dc_discipline, &sd).await; } loop { @@ -2320,6 +2392,7 @@ async fn rfm_async_loop( mut rfm: el6695_rt::rfm::Rfm, duration_sec: u64, dc_discipline: bool, + sd: ðercrab::SubDeviceRef<'_, ðercrab::SubDevice>, ) -> Result<(), Box> { use el6695_rt::rfm::*; println!("rfm-async: event-driven forwarding, no tick (Ctrl-C to stop)"); @@ -2388,6 +2461,7 @@ async fn rfm_async_loop( 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 = None; + let mut last_f640_print: u64 = 0; while !STOP.load(Ordering::Relaxed) { if duration_sec > 0 && start.elapsed().as_secs() >= duration_sec { @@ -2629,6 +2703,12 @@ async fn rfm_async_loop( } { let n = Instant::now(); seg_fprd_us = n.duration_since(cursor).as_micros() as u64; cursor = n; } + // F640 remote-write-cycle counter (independent of reap path) + if n_fpwr > 0 && n_fpwr % 512 == 0 && last_f640_print != n_fpwr { + last_f640_print = n_fpwr; + let c = sd.sdo_read::(0xF640, 1).await; + println!(" F640CNT n={} count={:?}", n_fpwr, c); + } // 6. reap FPWR completion if let Some(f) = pending_fpwr.as_mut() { if let Some(res) = futures_lite::future::poll_once(f.as_mut()).await { @@ -2639,6 +2719,7 @@ async fn rfm_async_loop( 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; }