ethercat-linux/src/main.rs

1696 lines
74 KiB
Rust
Raw Normal View History

//! J1900 <-> TwinCAT 1 kHz communication verification over an EL6695 bridge.
//!
//! SECONDARY-SIDE build: J1900 talks to the EL6695 secondary port. The
//! secondary side SM layout (probed from EEPROM) uses SM0/TX (0x1000, 1024 B,
//! ctrl=0x26) for J1900->TwinCAT data and SM1/RX (0x1600, 1024 B, ctrl=0x22)
//! for TwinCAT->J1900 data. Unlike the primary side there is NO external
//! timestamp object (0x1A02) — the data is raw 1024 B payload with no prefix.
//!
//! Two modes:
//! - `ticker` (default): echo TwinCAT's payload back on SM0 with the local
//! cycle number stamped at bytes 8..16.
//! - `probe`: read-only; dump the first bytes of SM1 every N cycles and do
//! NOT write to SM0 (avoids disturbing TwinCAT's view).
//!
//! Graceful SIGINT/SIGTERM shutdown with full statistics.
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};
use std::os::unix::io::AsRawFd;
use el6695_rt::{bridge, regs, rt, stats};
use ethercrab::{
std::{ethercat_now, tx_rx_task, tx_rx_task_io_uring},
MainDevice, MainDeviceConfig, PduStorage, Timeouts,
};
use stats::OnlineStats;
const MAX_SUBDEVICES: usize = 16;
const MAX_PDU_DATA: usize = 1100;
const MAX_FRAMES: usize = 16;
const PDI_LEN: usize = 3072;
// Secondary side SM layout (ESI-defined process data SMs)
const SM_TX_ADDR: u16 = 0x1C00; // SM2: outputs (J1900 writes data to TwinCAT)
const SM_TX_LEN: usize = 800;
const SM_RX_ADDR: u16 = 0x8E00; // SM3: inputs (J1900 reads data from TwinCAT)
// Full PDO: 2B diagnostic word + 8B payload (single-entry twin_layout blob
// expands 0x6000:01 to 8 bytes).
const SM_RX_LEN: usize = 800; // SM3 buffer (no diag word, pure 8B payload)
const SM_RX_PLD_LEN: usize = 800; // actual payload after stripping diagnostic word
const RX_PREFIX: usize = 0; // no diag word prefix (0x1C13=[0x1A08] only)
const PAYLOAD_LEN: usize = 800;
/// Abort after this many consecutive cycles with a TX or RX error.
const MAX_ERR_STREAK: u64 = 1000;
static STOP: AtomicBool = AtomicBool::new(false);
static PDU_STORAGE: std::sync::OnceLock<PduStorage<MAX_FRAMES, MAX_PDU_DATA>> =
std::sync::OnceLock::new();
extern "C" fn on_signal(_sig: libc::c_int) {
STOP.store(true, Ordering::Relaxed);
}
#[derive(Clone, Copy, PartialEq)]
enum Mode {
Ticker,
Probe,
Verify, // CoE audit: read 0xF008, layout hash, SM, identity
/// RFM ping: periodic reflective-memory write, wait for remote echo,
/// measure round-trip latency (same-clock RTT).
RfmPing,
/// RFM echo: poll REQ mailbox, copy to REP mailbox (remote ping peer).
RfmEcho,
/// RFM bridge: EL6695 RX -> RFM -> remote echo -> RFM -> EL6695 TX.
/// End-to-end CX<->x86 latency via reflective memory hop.
RfmBridge,
/// One-shot hardware ring-continuity check (prints ring=OK/BROKEN, exits).
RfmRing,
}
#[derive(Clone, Copy, Debug, PartialEq)]
enum Role {
/// Full PDO config owner: writes 0x1608/0x1A08/0x1C12/0x1C13.
/// Used when J1900 is the first/only master to configure the bridge.
ConfigOwner,
/// Read-only consumer: never writes PDO mapping/assignment.
/// Adopts whatever layout the primary-side owner (TwinCAT) left.
Consumer,
}
struct Config {
iface: String,
mode: Mode,
role: Role,
duration_sec: u64,
cycle_us: u64,
core: usize,
prio: u8,
quiet: bool,
/// Toggle an EL2202 digital output every cycle (oscilloscope jitter
/// waveform). Off by default: the outputs may drive real actuators.
el2202: bool,
/// Toggle BOTH EL2202 channels (0x03/0x00): output 1 -> EL1252 readback,
/// output 2 -> oscilloscope, same edges in two views.
el2202_dual: bool,
/// Read EL1252 DC latch timestamps every cycle (edge interval/phase
/// statistics; wire the signal under test to EL1252 channel 1).
el1252: bool,
/// Drive an EL2262 XFC output with DC-sync-locked edges (reference
/// square wave on the DC 1 ms grid).
el2262: bool,
/// Read the EL1252 every N cycles (frame-budget throttle when combined
/// with other per-cycle frames; edges are latched so nothing is lost
/// beyond time resolution).
el1252_every: u32,
/// Spin-lock mode: instead of a free-running ticker, poll FPRD until the
/// payload changes (new TwinCAT frame), then process immediately.
/// Phase-locks J1900 processing to the primary master's frame arrival.
spin: bool,
/// Enable DC SYNC0 on the EL6695 secondary (cycle = --cycle-us) and print
/// per-second DC phase/drift telemetry in the report line.
dc: bool,
/// Use the io_uring-based blocking TX/RX loop instead of AF_PACKET poll.
iouring: bool,
/// RFM device path override (e.g. /dev/rfm2g1) for multi-card hosts.
rfm_dev: Option<String>,
}
fn parse_args() -> Result<Config, String> {
let mut cfg = Config {
iface: "eth1".into(),
mode: Mode::Ticker,
role: Role::ConfigOwner,
duration_sec: 60,
cycle_us: 1000,
core: 2,
prio: 80,
quiet: false,
el2202: false,
el2202_dual: false,
el1252: false,
el2262: false,
el1252_every: 1,
spin: false,
dc: false,
iouring: false,
rfm_dev: None,
};
let mut args = std::env::args().skip(1).peekable();
if let Some(a) = args.peek() {
if !a.starts_with("--") {
cfg.iface = args.next().unwrap();
}
}
while let Some(a) = args.next() {
let mut val = |name: &str| -> Result<String, String> {
args.next().ok_or_else(|| format!("missing value for {}", name))
};
match a.as_str() {
"--mode" => {
cfg.mode = match val("--mode")?.as_str() {
"ticker" => Mode::Ticker,
"probe" => Mode::Probe,
"verify" => Mode::Verify,
"rfm-ping" => Mode::RfmPing,
"rfm-echo" => Mode::RfmEcho,
"rfm-bridge" => Mode::RfmBridge,
"rfm-ring" => Mode::RfmRing,
other => return Err(format!("unknown mode '{}'", other)),
}
}
"--duration" => {
cfg.duration_sec = val("--duration")?
.parse()
.map_err(|_| "bad --duration")?
}
"--cycle-us" => cfg.cycle_us = val("--cycle-us")?.parse().map_err(|_| "bad --cycle-us")?,
"--core" => cfg.core = val("--core")?.parse().map_err(|_| "bad --core")?,
"--prio" => cfg.prio = val("--prio")?.parse().map_err(|_| "bad --prio")?,
"--quiet" => cfg.quiet = true,
"--spin" => cfg.spin = true,
"--dc" => cfg.dc = true,
"--iouring" => cfg.iouring = true,
"--rfm-dev" => {
let n: u32 = val("--rfm-dev")?.parse().map_err(|_| "bad --rfm-dev (use card index, e.g. 0 or 1)")?;
cfg.rfm_dev = Some(format!("/dev/rfm2g{}", n));
}
"--el2202" => cfg.el2202 = true,
"--el2202-dual" => {
cfg.el2202 = true;
cfg.el2202_dual = true;
}
"--el1252" => cfg.el1252 = true,
"--el2262" => cfg.el2262 = true,
"--consumer" => cfg.role = Role::Consumer,
"--el1252-every" => {
cfg.el1252_every = val("--el1252-every")?.parse().map_err(|_| "bad --el1252-every")?
}
other => return Err(format!("unknown argument '{}'", other)),
}
}
if cfg.cycle_us == 0 {
return Err("--cycle-us must be > 0".into());
}
Ok(cfg)
}
fn main() {
let cfg = match parse_args() {
Ok(c) => c,
Err(e) => {
eprintln!("error: {}", e);
eprintln!(
"usage: el6695_rt [iface] [--mode ticker|probe] [--duration SEC] \
[--cycle-us N] [--core N] [--prio N] [--quiet] [--el2202] [--consumer]"
);
std::process::exit(2);
}
};
unsafe {
let mut sa: libc::sigaction = core::mem::zeroed();
sa.sa_sigaction = on_signal as usize;
sa.sa_flags = 0;
libc::sigemptyset(&mut sa.sa_mask);
libc::sigaction(libc::SIGINT, &sa, core::ptr::null_mut());
libc::sigaction(libc::SIGTERM, &sa, core::ptr::null_mut());
}
let code = match futures_lite::future::block_on(run(&cfg)) {
Ok(()) => 0,
Err(e) => {
eprintln!("Failed: {}", e);
1
}
};
std::process::exit(code);
}
async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
let cycle_ns = cfg.cycle_us * 1000;
println!("el6695-rt: J1900 <-> TwinCAT 1 kHz verification (EL6695 bridge, secondary side)");
println!(
" iface={} mode={} duration={}s cycle={}us core={} prio={}",
cfg.iface,
match cfg.mode {
Mode::Ticker => "ticker",
Mode::Probe => "probe",
Mode::Verify => "verify",
Mode::RfmPing => "rfm-ping",
Mode::RfmEcho => "rfm-echo",
Mode::RfmBridge => "rfm-bridge",
Mode::RfmRing => "rfm-ring",
},
cfg.duration_sec,
cfg.cycle_us,
cfg.core,
cfg.prio
);
let pdu_storage = PDU_STORAGE.get_or_init(PduStorage::new);
// ---- RFM-only modes: no EtherCAT bring-up needed ----
if matches!(cfg.mode, Mode::RfmPing | Mode::RfmEcho | Mode::RfmRing) {
if cfg.mode == Mode::RfmRing {
let rfm = el6695_rt::rfm::Rfm::open_path(cfg.rfm_dev.as_deref())?;
let fd = rfm.device_fd.as_raw_fd();
let ok = el6695_rt::rfm::ring_cont(fd)?;
println!("ring={}", if ok { "OK" } else { "BROKEN" });
std::process::exit(if ok { 0 } else { 1 });
}
return rfm_mode_main(cfg).await;
}
let (tx, rx, pdu_loop) = pdu_storage.try_split().map_err(|_| "PDU storage already in use")?;
let maindevice = MainDevice::new(
pdu_loop,
Timeouts {
state_transition: Duration::from_millis(10000),
mailbox_response: Duration::from_millis(2000),
eeprom: Duration::from_millis(500),
..Default::default()
},
MainDeviceConfig::default(),
);
let iface = cfg.iface.clone();
let use_iouring = cfg.iouring;
let _txrx = std::thread::spawn(move || {
if let Err(e) = rt::promote_current_thread(1, 79, false) {
eprintln!("tx_rx RT promotion failed (continuing): {}", e);
}
if use_iouring {
match tx_rx_task_io_uring(&iface, tx, rx) {
Ok(_) => {}
Err(e) => eprintln!("io_uring tx_rx task ended with error: {:?}", e),
}
} else {
let task = tx_rx_task(&iface, tx, rx).expect("create tx_rx task");
if let Err(e) = futures_lite::future::block_on(task) {
eprintln!("tx_rx task ended with error: {:?}", e);
}
}
});
println!("Scanning segment (INIT -> PREOP)...");
let group = maindevice
.init_single_group::<MAX_SUBDEVICES, PDI_LEN>(ethercat_now)
.await?;
let mut el6695_index = None;
for (idx, subdevice) in group.iter(&maindevice).enumerate() {
if subdevice.name().contains("EL6695") {
el6695_index = Some(idx);
println!("[{}] Found EL6695", idx);
}
}
let el6695_index = el6695_index.ok_or("EL6695 not found on bus")?;
let sd = group.subdevice(&maindevice, el6695_index)?;
let station = sd.configured_address();
println!("EL6695 station address: 0x{:04X}", station);
// ---- Probe SM registers ----
async fn read_sm(md: &MainDevice<'_>, st: u16, n: u8) -> (u16, u16, u8, u8) {
let start = ethercrab::Command::fprd(st, regs::sm_base(n as u16) + regs::SM_OFF_START).receive::<u16>(md).await.unwrap_or(0xFFFF);
let len = ethercrab::Command::fprd(st, regs::sm_base(n as u16) + regs::SM_OFF_LEN).receive::<u16>(md).await.unwrap_or(0xFFFF);
let ctrl = ethercrab::Command::fprd(st, regs::sm_base(n as u16) + regs::SM_OFF_CTRL).receive::<u8>(md).await.unwrap_or(0xFF);
let ena = ethercrab::Command::fprd(st, regs::sm_base(n as u16) + regs::SM_OFF_ENABLE).receive::<u8>(md).await.unwrap_or(0xFF);
(start, len, ctrl, ena)
}
for n in [0u8, 1, 2, 3] {
let (s, l, c, e) = read_sm(&maindevice, station, n).await;
println!(" SM{}: start=0x{:04X} len={} ctrl=0x{:02X} enable=0x{:02X}", n, s, l, c, e);
}
// Probe FMMU registers
for n in 0u16..4 {
let reg = 0x0600 + n * 16;
match ethercrab::Command::fprd(station, reg)
.receive_slice(&maindevice, 16)
.await
{
Ok(pdu) => {
let raw: &[u8] = &pdu;
let logical_start = u32::from_le_bytes(raw[0..4].try_into().unwrap());
let length = u16::from_le_bytes(raw[4..6].try_into().unwrap());
let phy_start = u16::from_le_bytes(raw[8..10].try_into().unwrap());
println!(" FMMU{}: logical_start=0x{:08X} len={} physical=0x{:04X} enable=0x{:02X}", n, logical_start, length, phy_start, raw[12]);
}
Err(e) => println!(" FMMU{} read error: {:?}", n, e),
}
}
// ---- Configure SM2/SM3 for process data (SM0/SM1 are EEPROM-fixed mailbox) ----
// 0x1C12/0x1C13 exist (empty), 0x1C10/0x1C11 do not.
// SM2/SM3 registers are writable because they are not defined in EEPROM.
println!("\nTesting SM0/SM1 data access...");
// Read SM1 (buffered input, 0x1600)
match ethercrab::Command::fprd(station, 0x1600)
.receive_slice(&maindevice, 32)
.await
{
Ok(pdu) => {
let raw: &[u8] = &pdu;
println!(" SM1 FPRD(0x1600,32) OK: {:02X?}", &raw[..32.min(raw.len())]);
}
Err(e) => println!(" SM1 FPRD error: {:?}", e),
}
// Write SM0 (mailbox, 0x1000)
match ethercrab::Command::fpwr(station, 0x1000)
.send(&maindevice, &[0xBBu8; 16])
.await
{
Ok(_) => println!(" SM0 FPWR(0x1000,16) OK"),
Err(e) => println!(" SM0 FPWR error: {:?}", e),
}
// Read back SM0
match ethercrab::Command::fprd(station, 0x1000)
.receive_slice(&maindevice, 16)
.await
{
Ok(pdu) => {
let raw: &[u8] = &pdu;
println!(" SM0 FPRD(0x1000,16) back: {:02X?}", &raw[..16.min(raw.len())]);
}
Err(e) => println!(" SM0 FPRD back error: {:?}", e),
}
// Remove SM2/SM3 config (was causing SAFEOP 0x001D)
// Safety: clear SM2/SM3 in case previous runs left stale values
let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_ENABLE)
.send(&maindevice, 0x00u8).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_ENABLE)
.send(&maindevice, 0x00u8).await;
// ---- Verify mode: CoE audit (then exit, no SAFEOP) ----
if cfg.mode == Mode::Verify {
println!("\n=== CoE Audit ===");
// 0xF008: NoCoeStorage / vendor reset
let f008_0 = sd.sdo_read::<u32>(0xF008, 0).await;
let f008_1 = sd.sdo_read::<u8>(0xF008, 1).await;
println!(" 0xF008:00 = {:?} sub1={:?}", f008_0, f008_1);
let no_coe_storage = f008_0.unwrap_or(0) == 0x12345678;
println!(" NoCoeStorage: {}", if no_coe_storage { "⚠ ENABLED (0x12345678) — config will NOT persist to EEPROM" } else { "✓ disabled" });
// 0x1018: Identity
for sub in 1..=4u8 {
let v = sd.sdo_read::<u32>(0x1018, sub).await;
let label = match sub { 1 => "Vendor", 2 => "Product", 3 => "Revision", _ => "Serial" };
println!(" 0x1018:{:02} ({}) = 0x{:08X}", sub, label, v.unwrap_or(0xFFFFFFFF));
}
// Full PDO layout
let snap = bridge::snapshot_layout(&sd, &maindevice, station).await;
let h = bridge::compute_layout_hash(&snap);
println!(" 0x1C12 assign: {:?}", snap.tx_assign);
println!(" 0x1C13 assign: {:?}", snap.rx_assign);
println!(" 0x1608 entries: {} (total {} bits)", snap.tx_entries.len(), snap.tx_total_bits);
if !snap.tx_entries.is_empty() {
for (i, (idx, sub, bl)) in snap.tx_entries.iter().enumerate().take(3) {
println!(" [{:02}] 0x{:04X}:{:02} @ {} bits", i+1, idx, sub, bl);
}
if snap.tx_entries.len() > 3 {
println!(" ... ({} more)", snap.tx_entries.len() - 3);
}
}
println!(" 0x1A08 entries: {} (total {} bits)", snap.rx_entries.len(), snap.rx_total_bits);
if !snap.rx_entries.is_empty() {
for (i, (idx, sub, bl)) in snap.rx_entries.iter().enumerate().take(3) {
println!(" [{:02}] 0x{:04X}:{:02} @ {} bits", i+1, idx, sub, bl);
}
if snap.rx_entries.len() > 3 {
println!(" ... ({} more)", snap.rx_entries.len() - 3);
}
}
println!(" SM2 len={} SM3 len={}", snap.sm2_len, snap.sm3_len);
println!(" Layout CRC32 = 0x{:08X}", h);
println!(" Expected hash = 0x{:08X}", bridge::EXPECTED_LAYOUT_HASH.load(std::sync::atomic::Ordering::Relaxed));
println!(" 0x6000[0] = {}", sd.sdo_read::<u8>(0x6000, 0).await.unwrap_or(0xFF));
println!(" 0x7000[0] = {}", sd.sdo_read::<u8>(0x7000, 0).await.unwrap_or(0xFF));
// AL state / error
let (al, _) = bridge::drive_state(&maindevice, station, 0x0002).await;
println!(" AL=0x{:04X} err={}", al, al & regs::AL_ERROR != 0);
println!("=== CoE Audit complete ===");
return Ok(());
}
// ---- ensure_link: adopt-if-valid (Consumer) or full reconfigure (ConfigOwner) ----
println!("\nensure_link: role={:?} ...", cfg.role);
// Read current PDO layout from the bridge
let layout_ok = bridge::layout_matches(&sd, &maindevice, station).await.unwrap_or(false);
let (al, alc) = bridge::drive_state(&maindevice, station, 0x0002).await;
let has_err = al & regs::AL_ERROR != 0;
let al_state = al & 0x0F;
println!(" AL=0x{:04X} err={} layout_ok={}", al, has_err, layout_ok);
match cfg.role {
Role::Consumer => {
// ---- Consumer: idempotent writes, no INIT→PREOP cycle ----
// SM/PDO config must ALWAYS be written after scan (scanning wipes
// SM registers and PDO assign/map). The writes are idempotent:
// same layout as the expected expanded 1024-byte config.
// Clear any PREOP+ERR (L1 recovery)
if has_err {
println!(" Consumer: clearing PREOP+ERR ...");
let _ = ethercrab::Command::fpwr(station, regs::AL_CONTROL)
.send(&maindevice, 0x0002u16).await;
std::thread::sleep(Duration::from_millis(300));
}
// Skip INIT→PREOP — preserves TwinCAT's persisted CoE objects.
println!(" Consumer: idempotent SM/FMMU/PDO config");
// SM2/SM3 registers
let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_ENABLE)
.send(&maindevice, 0x00u8).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_START)
.send(&maindevice, 0x1C00u16).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_LEN)
.send(&maindevice, (PAYLOAD_LEN as u16)).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_CTRL)
.send(&maindevice, 0x64u8).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_ENABLE)
.send(&maindevice, 0x01u8).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_ENABLE)
.send(&maindevice, 0x00u8).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_START)
.send(&maindevice, 0x8E00u16).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_LEN)
.send(&maindevice, (PAYLOAD_LEN as u16)).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_CTRL)
.send(&maindevice, 0x20u8).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_ENABLE)
.send(&maindevice, 0x01u8).await;
bridge::write_fmmus_secondary(&maindevice, station, PAYLOAD_LEN as u16, PAYLOAD_LEN as u16).await;
// PDO maps and assigns (whitelisted, idempotent)
println!(" Consumer: write PDO mappings...");
let blob = bridge::twin_layout_blob_sized(0x7000, PAYLOAD_LEN);
let _ = bridge::sdo_write_ca_checked(&sd, 0x1608, 0, &blob).await;
let blob = bridge::twin_layout_blob_sized(0x6000, PAYLOAD_LEN);
let _ = bridge::sdo_write_ca_checked(&sd, 0x1A08, 0, &blob).await;
println!(" Consumer: write PDO assignments...");
let blob = bridge::assign_blob(&[0x1608]);
let _ = bridge::sdo_write_ca_checked(&sd, 0x1C12, 0, &blob).await;
let blob = bridge::assign_blob(&[0x1A08]);
let _ = bridge::sdo_write_ca_checked(&sd, 0x1C13, 0, &blob).await;
let _o7 = sd.sdo_read::<u8>(0x7000, 0).await.unwrap_or(0xFF);
let _o6 = sd.sdo_read::<u8>(0x6000, 0).await.unwrap_or(0xFF);
let snap = bridge::snapshot_layout(&sd, &maindevice, station).await;
let h = bridge::compute_layout_hash(&snap);
println!(" Consumer: dyn 0x6000[0]={} 0x7000[0]={} hash=0x{:08X}", _o6, _o7, h);
}
Role::ConfigOwner => {
// ---- ConfigOwner: full INIT→PREOP + PDO rebuild ----
// Clear any PREOP+ERR by re-requesting PREOP
if has_err {
println!(" ConfigOwner: clearing PREOP+ERR ...");
let _ = ethercrab::Command::fpwr(station, regs::AL_CONTROL)
.send(&maindevice, 0x0002u16).await;
std::thread::sleep(Duration::from_millis(300));
}
// INIT→PREOP cycle to reset CoE state
println!(" INIT→PREOP cycle...");
let (al_i, _) = bridge::drive_state(&maindevice, station, 0x0001).await;
println!(" INIT: {}", bridge::verdict(al_i, 0));
if al_i & 0x0F != 0x01 {
return Err(format!("INIT failed: {}", bridge::verdict(al_i, 0)).into());
}
let (al_p, alc_p) = bridge::drive_state(&maindevice, station, 0x0002).await;
println!(" PREOP: {}", bridge::verdict(al_p, alc_p));
if al_p & 0x0F != 0x02 {
return Err(format!("PREOP failed: {}", bridge::verdict(al_p, alc_p)).into());
}
// Write PDO mappings (0x1608 TxPDO, 0x1A08 RxPDO)
// Whitelist-checked: only {0x1608,0x1A08,0x1C12,0x1C13,0xF008,0xF800} allowed.
println!(" Write PDO mappings...");
let blob = bridge::twin_layout_blob_sized(0x7000, PAYLOAD_LEN);
let _ = bridge::sdo_write_ca_checked(&sd, 0x1608, 0, &blob).await;
let blob = bridge::twin_layout_blob_sized(0x6000, PAYLOAD_LEN);
let _ = bridge::sdo_write_ca_checked(&sd, 0x1A08, 0, &blob).await;
// Write PDO assignments (triggers 0x6000/0x7000 object creation)
println!(" Write PDO assignments...");
let blob = bridge::assign_blob(&[0x1608]);
let _ = bridge::sdo_write_ca_checked(&sd, 0x1C12, 0, &blob).await;
let blob = bridge::assign_blob(&[0x1A08]);
let _ = bridge::sdo_write_ca_checked(&sd, 0x1C13, 0, &blob).await;
// Verify the dynamic objects were instantiated
let _o7 = sd.sdo_read::<u8>(0x7000, 0).await.unwrap_or(0xFF);
let _o6 = sd.sdo_read::<u8>(0x6000, 0).await.unwrap_or(0xFF);
// Compute layout hash after successful config
let snap = bridge::snapshot_layout(&sd, &maindevice, station).await;
let h = bridge::compute_layout_hash(&snap);
println!(" dyn: 0x6000[0]={} 0x7000[0]={} layout_hash=0x{:08X}", _o6, _o7, h);
// Configure SM2/SM3 + FMMU
let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_ENABLE)
.send(&maindevice, 0x00u8).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_START)
.send(&maindevice, 0x1C00u16).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_LEN)
.send(&maindevice, (PAYLOAD_LEN as u16)).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_CTRL)
.send(&maindevice, 0x64u8).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_ENABLE)
.send(&maindevice, 0x01u8).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_ENABLE)
.send(&maindevice, 0x00u8).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_START)
.send(&maindevice, 0x8E00u16).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_LEN)
.send(&maindevice, (PAYLOAD_LEN as u16)).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_CTRL)
.send(&maindevice, 0x20u8).await;
let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_ENABLE)
.send(&maindevice, 0x01u8).await;
bridge::write_fmmus_secondary(&maindevice, station, PAYLOAD_LEN as u16, PAYLOAD_LEN as u16).await;
}
}
// SAFEOP
let (al, alc) = bridge::drive_state(&maindevice, station, 0x0004).await;
println!(" SAFEOP: {}", bridge::verdict(al, alc));
if al & 0x0F != 0x04 || al & 0x10 != 0 {
return Err(format!("Cannot reach SAFEOP: {}", bridge::verdict(al, alc)).into());
}
println!(" >>> SAFEOP achieved! <<<");
let (al2, alc2) = bridge::drive_state(&maindevice, station, 0x0008).await;
println!(" OP: {}", bridge::verdict(al2, alc2));
if al2 & 0x0F != 0x08 || al2 & 0x10 != 0 {
return Err(format!("OP rejected: {}", bridge::verdict(al2, alc2)).into());
}
println!(" >>> OP achieved! <<<");
// ---- Optional DC SYNC0 on the EL6695 secondary ----
// ethercrab init already distributed the system time (offset/propagation
// delay/static drift). Here we only start the SYNC0 unit on --cycle-us
// grid and verify the clock actually advances.
let mut dc_ref: Option<(u64, u64)> = None; // (dc_time, mono_time) at setup
if cfg.dc {
let t1 = ethercrab::Command::fprd(station, regs::DC_SYSTEM_TIME)
.receive::<u64>(&maindevice)
.await
.unwrap_or(0);
std::thread::sleep(Duration::from_millis(100));
let t2 = ethercrab::Command::fprd(station, regs::DC_SYSTEM_TIME)
.receive::<u64>(&maindevice)
.await
.unwrap_or(0);
let adv = t2.wrapping_sub(t1);
println!("DC system time: {} -> {} (adv {} ns in 100ms)", t1, t2, adv);
if adv < 50_000_000 || adv > 500_000_000 {
println!("DC: clock not advancing sanely; SYNC0 setup skipped");
} else {
// Diagnostics: which DC register groups actually exist?
for (reg, name) in [
(0x0900u16, "rx_time_port0(u32)"),
(0x0920u16, "sys_time_offset(u64)"),
(0x0928u16, "prop_delay(u32)"),
(0x0930u16, "drift... (u16)"),
(0x0980u16, "sync_activation(u8)"),
(0x0981u16, "sync_active_byte(u8)"),
(0x09A0u16, "sync0_cycle(u32)"),
(0x09AEu16, "sync0_status(u8)"),
] {
match ethercrab::Command::fprd(station, reg)
.receive_slice(&maindevice, 8)
.await
{
Ok(pdu) => {
let raw: &[u8] = &pdu;
println!(" DC probe 0x{:04X} {}: {:02X?}", reg, name, &raw[..8.min(raw.len())]);
}
Err(e) => println!(" DC probe 0x{:04X} {}: ERR {:?}", reg, name, e),
}
}
// CoE view: 0x1C32 (SM2 outputs) / 0x1C33 (SM3 inputs) sync params
for idx in [0x1C32u16, 0x1C33] {
let n = sd.sdo_read::<u8>(idx, 0).await.unwrap_or(0xFF);
println!(" CoE 0x{:04X}: {} subentries", idx, n);
for (sub, name) in [(0x01u8, "sync_mode"), (0x04, "modes_supported"), (0x0B, "sm_event_missed"), (0x0C, "cycle_exceeded")] {
match sd.sdo_read::<u16>(idx, sub).await {
Ok(x) => println!(" CoE 0x{:04X}:{:02X} {} = 0x{:04X}", idx, sub, name, x),
Err(e) => println!(" CoE 0x{:04X}:{:02X} {} ERR {:?}", idx, sub, name, e),
}
}
match sd.sdo_read::<u32>(idx, 0x02).await {
Ok(x) => println!(" CoE 0x{:04X}:02 cycle_time_ns = {}", idx, x),
Err(e) => println!(" CoE 0x{:04X}:02 cycle_time_ns ERR {:?}", idx, e),
}
match sd.sdo_read::<u8>(idx, 0x20).await {
Ok(x) => println!(" CoE 0x{:04X}:20 sync_error = {}", idx, x),
Err(e) => println!(" CoE 0x{:04X}:20 sync_error ERR {:?}", idx, e),
}
}
let start = t2.wrapping_add(100_000_000);
let w1 = ethercrab::Command::fpwr(station, regs::DC_SYNC_START_TIME)
.send(&maindevice, start)
.await;
println!(" DC write start_time(0x0990): {:?}", w1.as_ref().map(|_| "OK").map_err(|e| format!("{:?}", e)));
let w2 = ethercrab::Command::fpwr(station, regs::DC_SYNC0_CYCLE)
.send(&maindevice, cycle_ns as u32)
.await;
println!(" DC write sync0_cycle(0x09A0): {:?}", w2.as_ref().map(|_| "OK").map_err(|e| format!("{:?}", e)));
let w3 = regs::rmw_u8(
&maindevice,
station,
regs::DC_SYNC_ACTIVATION,
regs::DC_SYNC0_ACTIVATE,
0,
)
.await;
println!(" DC rmw activation(0x0980): {:?}", w3.as_ref().map(|v| format!("0x{:02X}", v)).map_err(|e| format!("{:?}", e)));
// rmw needs a read (fails here); try a blind full-byte write instead
let w4 = ethercrab::Command::fpwr(station, regs::DC_SYNC_ACTIVATION)
.send(&maindevice, regs::DC_SYNC0_ACTIVATE)
.await;
println!(" DC blind write activation(0x0980)=0x01: {:?}", w4.as_ref().map(|_| "OK").map_err(|e| format!("{:?}", e)));
std::thread::sleep(Duration::from_millis(150));
// Try selecting DC-Synchron SYNC0 via firmware object (mode 2).
// May be rejected in OP; worth observing the abort code either way.
let mut coe_dc_ok = true;
for idx in [0x1C32u16, 0x1C33] {
match sd.sdo_write::<u16>(idx, 0x01, 2).await {
Ok(_) => println!(" CoE write 0x{:04X}:01 sync_mode=2 OK", idx),
Err(e) => {
coe_dc_ok = false;
println!(" CoE write 0x{:04X}:01 sync_mode=2 ERR {:?}", idx, e);
}
}
match sd.sdo_read::<u16>(idx, 0x01).await {
Ok(x) => println!(" CoE-after 0x{:04X}:01 sync_mode = {}", idx, x),
Err(e) => {
coe_dc_ok = false;
println!(" CoE-after 0x{:04X}:01 sync_mode ERR {:?}", idx, e);
}
}
}
if coe_dc_ok {
let t_dc = ethercrab::Command::fprd(station, regs::DC_SYSTEM_TIME)
.receive::<u64>(&maindevice)
.await
.unwrap_or(0);
dc_ref = Some((t_dc, rt::now_ns()));
println!("DC-Synchron: firmware sync_mode=2 accepted, DC telemetry on");
}
let act = ethercrab::Command::fprd(station, regs::DC_SYNC_ACTIVATION)
.receive::<u8>(&maindevice)
.await;
let cyc0 = ethercrab::Command::fprd(station, regs::DC_SYNC0_CYCLE)
.receive::<u32>(&maindevice)
.await;
let t3 = ethercrab::Command::fprd(station, regs::DC_SYSTEM_TIME)
.receive::<u64>(&maindevice)
.await
.unwrap_or(0);
let (al, alc) = bridge::drive_state(&maindevice, station, 0x0008).await;
println!(" AL after DC writes: {}", bridge::verdict(al, alc));
println!(
"DC SYNC0: act={:?} cycle={:?} start=+100ms (t3={} start_delta={}ns)",
act.as_ref().map(|v| format!("0x{:02X}", v)).map_err(|e| format!("{:?}", e)),
cyc0.as_ref().map_err(|e| format!("{:?}", e)),
t3,
t3.wrapping_sub(start) as i64
);
if matches!(act, Ok(a) if a & regs::DC_SYNC0_ACTIVATE != 0)
&& matches!(cyc0, Ok(c) if c == cycle_ns as u32)
{
dc_ref = Some((t3, rt::now_ns()));
println!("DC SYNC0: active on EL6695 secondary");
} else {
println!("DC SYNC0: readback mismatch; running without DC telemetry");
}
}
}
// ---- Optional EL2202 oscilloscope waveform (toggle one DO per cycle) ----
const EL2202_OUT_ADDR: u16 = 0x0F00;
let mut el2202: Option<(u16, u16)> = None;
if cfg.el2202 {
let mut found = false;
for (idx, subdevice) in group.iter(&maindevice).enumerate() {
if subdevice.name().contains("EL2202") {
found = true;
let sd2 = group.subdevice(&maindevice, idx)?;
let st2 = sd2.configured_address();
println!("[{}] Found EL2202, station 0x{:04X}", idx, st2);
let _ = ethercrab::Command::fpwr(st2, regs::sm_base(0) + regs::SM_OFF_START)
.send(&maindevice, EL2202_OUT_ADDR)
.await;
let _ = ethercrab::Command::fpwr(st2, regs::sm_base(0) + regs::SM_OFF_LEN)
.send(&maindevice, 1u16)
.await;
let _ = ethercrab::Command::fpwr(st2, regs::sm_base(0) + regs::SM_OFF_CTRL)
.send(&maindevice, regs::SM_CTRL_EL2202_OUTPUTS)
.await;
let _ = regs::rmw_u8(
&maindevice,
st2,
regs::sm_base(0) + regs::SM_OFF_ENABLE,
regs::SM_ENABLE | regs::SM_ENABLE_FAMILY,
0,
)
.await;
let (al, alc) = bridge::drive_state(&maindevice, st2, regs::AL_SAFEOP).await;
println!("EL2202 SAFEOP: {}", bridge::verdict(al, alc));
let (al, alc) = bridge::drive_state(&maindevice, st2, regs::AL_OP).await;
println!("EL2202 OP: {}", bridge::verdict(al, alc));
if al & regs::AL_STATE_MASK != regs::AL_OP || al & regs::AL_ERROR != 0 {
return Err(format!("EL2202 OP rejected: {}", bridge::verdict(al, alc)).into());
}
let _ = ethercrab::Command::fpwr(st2, EL2202_OUT_ADDR)
.send(&maindevice, regs::EL2202_CH1_DATA)
.await;
let rb = ethercrab::Command::fprd(st2, EL2202_OUT_ADDR)
.receive::<u8>(&maindevice)
.await
.unwrap_or(0xFF);
if rb & regs::EL2202_CH1_DATA == regs::EL2202_CH1_DATA {
println!(
"EL2202 scope output: verified, toggle 1 byte @0x{:04X} every cycle",
EL2202_OUT_ADDR
);
let _ = ethercrab::Command::fpwr(st2, EL2202_OUT_ADDR)
.send(&maindevice, 0x00u8)
.await;
el2202 = Some((st2, EL2202_OUT_ADDR));
} else {
println!("EL2202 output readback failed (0x{:02X}); scope output disabled", rb);
}
}
}
if !found {
println!("--el2202 given but no EL2202 on bus; scope output disabled");
}
if found && el2202.is_none() {
println!("EL2202 present but output setup failed; scope output disabled");
}
}
// ---- Optional EL2262 DC-locked reference output ----
const EL2262_OUT_ADDR: u16 = 0x1000;
let mut el2262: Option<u16> = None;
if cfg.el2262 {
let mut found = false;
for (idx, subdevice) in group.iter(&maindevice).enumerate() {
if subdevice.name().contains("EL2262") {
found = true;
let sd3 = group.subdevice(&maindevice, idx)?;
let st3 = sd3.configured_address();
println!("[{}] Found EL2262, station 0x{:04X}", idx, st3);
let t = ethercrab::Command::fprd(st3, regs::DC_SYSTEM_TIME)
.receive::<u64>(&maindevice)
.await
.unwrap_or(0);
let start = t.wrapping_add(100_000_000);
let _ = ethercrab::Command::fpwr(st3, regs::DC_SYNC_START_TIME)
.send(&maindevice, start)
.await;
let _ = ethercrab::Command::fpwr(st3, regs::DC_SYNC0_CYCLE)
.send(&maindevice, cycle_ns as u32)
.await;
let _ = ethercrab::Command::fpwr(st3, regs::DC_SYNC1_CYCLE)
.send(&maindevice, cycle_ns as u32)
.await;
let _ = regs::rmw_u8(
&maindevice,
st3,
regs::DC_SYNC_ACTIVATION,
regs::DC_SYNC0_ACTIVATE | regs::DC_SYNC1_ACTIVATE,
0,
)
.await;
println!("EL2262 DC sync: sync0/1 cycle={}ns start=+100ms (t={})", cycle_ns, t);
let _ = ethercrab::Command::fpwr(st3, regs::sm_base(0) + regs::SM_OFF_START)
.send(&maindevice, EL2262_OUT_ADDR)
.await;
let _ = ethercrab::Command::fpwr(st3, regs::sm_base(0) + regs::SM_OFF_LEN)
.send(&maindevice, 1u16)
.await;
let _ = ethercrab::Command::fpwr(st3, regs::sm_base(0) + regs::SM_OFF_CTRL)
.send(&maindevice, regs::SM_CTRL_OUTPUTS_ESI)
.await;
let _ = regs::rmw_u8(
&maindevice,
st3,
regs::sm_base(0) + regs::SM_OFF_ENABLE,
regs::SM_ENABLE | regs::SM_ENABLE_FAMILY,
0,
)
.await;
let (al, alc) = bridge::drive_state(&maindevice, st3, regs::AL_SAFEOP).await;
println!("EL2262 SAFEOP: {}", bridge::verdict(al, alc));
let (al, alc) = bridge::drive_state(&maindevice, st3, regs::AL_OP).await;
println!("EL2262 OP: {}", bridge::verdict(al, alc));
if al & regs::AL_STATE_MASK != regs::AL_OP || al & regs::AL_ERROR != 0 {
return Err(format!("EL2262 OP rejected: {}", bridge::verdict(al, alc)).into());
}
el2262 = Some(st3);
}
}
if !found {
println!("--el2262 given but no EL2262 on bus; reference output disabled");
}
}
// ---- Optional EL1252 DC latch timestamp readback ----
let mut el1252: Option<u16> = None;
if cfg.el1252 {
let mut found = false;
for (idx, subdevice) in group.iter(&maindevice).enumerate() {
if subdevice.name().contains("EL1252") {
found = true;
let sd4 = group.subdevice(&maindevice, idx)?;
let st4 = sd4.configured_address();
println!("[{}] Found EL1252, station 0x{:04X} (latch readback)", idx, st4);
let _ = ethercrab::Command::fpwr(st4, regs::sm_base(0) + regs::SM_OFF_START)
.send(&maindevice, 0x1000u16)
.await;
let _ = ethercrab::Command::fpwr(st4, regs::sm_base(0) + regs::SM_OFF_LEN)
.send(&maindevice, 1u16)
.await;
let _ = ethercrab::Command::fpwr(st4, regs::sm_base(0) + regs::SM_OFF_CTRL)
.send(&maindevice, regs::SM_CTRL_EL1252_INPUTS)
.await;
let _ = regs::rmw_u8(
&maindevice,
st4,
regs::sm_base(0) + regs::SM_OFF_ENABLE,
regs::SM_ENABLE,
0,
)
.await;
let (al, alc) = bridge::drive_state(&maindevice, st4, regs::AL_SAFEOP).await;
println!("EL1252 SAFEOP: {}", bridge::verdict(al, alc));
let (al, alc) = bridge::drive_state(&maindevice, st4, regs::AL_OP).await;
println!("EL1252 OP: {}", bridge::verdict(al, alc));
if al & regs::AL_STATE_MASK != regs::AL_OP || al & regs::AL_ERROR != 0 {
return Err(format!("EL1252 OP rejected: {}", bridge::verdict(al, alc)).into());
}
el1252 = Some(st4);
}
}
if !found {
println!("--el1252 given but no EL1252 on bus; latch readback disabled");
}
}
// ---- Pre-allocate everything, then promote to RT ----
let mut tx_buf = vec![0u8; SM_TX_LEN];
let mut rx_buf = vec![0u8; SM_RX_LEN];
let mut jitter_st = OnlineStats::new("period_jitter", 100, -5_000_000, 5_000_000);
let mut exec_st = OnlineStats::new("exec_time", 1_000, 0, 20_000_000);
let mut rx_rtt_st = OnlineStats::new("fprd_rtt", 1_000, 0, 20_000_000);
let mut tx_rtt_st = OnlineStats::new("fpwr_rtt", 1_000, 0, 20_000_000);
println!("Promoting to RT: core={}, fifo={}, mlock=true", cfg.core, cfg.prio);
if let Err(e) = rt::promote_current_thread(cfg.core, cfg.prio, true) {
eprintln!("RT promotion failed (needs root, continuing anyway): {}", e);
}
rt::prefault_stack(256 * 1024);
// ---- Real-time data loop ----
println!(
"\n>>> running: {} loop, echo payload, stats (Ctrl-C for graceful stop)",
match cfg.mode {
Mode::Ticker => "ticker",
Mode::Probe => "probe (read-only, no echo)",
Mode::Verify => "verify (CoE audit, no data loop)",
Mode::RfmBridge => "rfm-bridge (EL6695 <-> RFM <-> remote)",
Mode::RfmPing | Mode::RfmEcho => unreachable!(),
}
);
let start = Instant::now();
let mut ticker = rt::Ticker::new(cycle_ns);
let mut cyc: u64 = 0;
let mut tx_err: u64 = 0;
let mut rx_err: u64 = 0;
let mut err_streak: u64 = 0;
let mut no_tc_seq: u64 = 0;
let mut el2202_err: u64 = 0;
let mut el2262_err: u64 = 0;
let mut el1252_err: u64 = 0;
let mut latch_pos0: u64 = 0;
let mut latch_neg0: u64 = 0;
let mut latch_pos1: u64 = 0;
let mut latch_neg1: u64 = 0;
let mut pos_edges: u64 = 0;
let mut neg_edges: u64 = 0;
let mut pos1_edges: u64 = 0;
let mut el1252_st: [u8; 2] = [0, 0];
let mut el1252_in: u8 = 0;
let mut latch_interval_st = OnlineStats::new("el1252_ch1_edge_interval", 1_000, 0, 20_000_000);
let mut latch_phase_st = OnlineStats::new("el1252_ch1_edge_phase", 100, 0, 1_000_000);
let mut latch1_interval_st = OnlineStats::new("el1252_ch2_edge_interval", 1_000, 0, 20_000_000);
let mut latch1_phase_st = OnlineStats::new("el1252_ch2_edge_phase", 100, 0, 1_000_000);
let mut last_report = Instant::now();
// J1900 data verification counters
let mut prev_tc_seq: u64 = 0;
let mut rx_stale: u64 = 0; // tc_seq unchanged (stale read)
let mut rx_nonmono: u64 = 0; // tc_seq decreased (non-monotonic)
let mut comp_err: u64 = 0; // internal computation errors
let mut payload_checked: u64 = 0; // J1900 payload verification cycles
let mut payload_mismatch: u64 = 0; // J1900 payload mismatch bytes accumulated
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; PAYLOAD_LEN]));
let tx_slot_b = Box::leak(Box::new([0u8; PAYLOAD_LEN]));
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;
// RFM bridge context (only opened in rfm-bridge mode).
let mut rfm_ctx: Option<el6695_rt::rfm::Rfm> = if cfg.mode == Mode::RfmBridge {
let r = el6695_rt::rfm::Rfm::open_path(cfg.rfm_dev.as_deref())?;
println!(
"rfm-bridge: BAR2 reflective mem = {} MB, node_id={}",
r.mem.len() / (1024 * 1024),
r.node_id
);
Some(r)
} else {
None
};
let mut bridge_req_seq: u64 = 0;
let mut bridge_ok: u64 = 0;
let mut bridge_timeouts: u64 = 0;
let mut bridge_rtt_st = OnlineStats::new("rfm_hop_rtt", 100, 0, 20_000_000);
loop {
if STOP.load(Ordering::Relaxed) {
println!("\n(stop requested, finishing cycle {})", cyc);
break;
}
if cfg.duration_sec > 0 && start.elapsed().as_secs() >= cfg.duration_sec {
break;
}
let mut rx_ok = true;
let spin_mode = cfg.spin && cfg.mode == Mode::Ticker;
let mut jitter_ns: i64 = 0;
if spin_mode {
// ---- Spin-lock: poll FPRD until the payload changes ----
// Phase-locks processing to TwinCAT's frame arrival instead of a
// free-running ticker. Timeout at 1.5 periods keeps the loop
// alive (counted as stale) if the primary master stops.
let spin_t0 = Instant::now();
let timeout_ns = cycle_ns + cycle_ns / 2;
let mut fresh = false;
loop {
match ethercrab::Command::fprd(station, SM_RX_ADDR)
.receive_slice(&maindevice, SM_RX_LEN as u16)
.await
{
Ok(pdu) => {
let raw: &[u8] = &pdu;
let n = raw.len().min(SM_RX_LEN);
if n > RX_PREFIX {
let copy_len = (n - RX_PREFIX).min(SM_RX_PLD_LEN);
rx_buf[..copy_len].copy_from_slice(&raw[RX_PREFIX..RX_PREFIX + copy_len]);
}
if rx_buf[..8] != prev_rx8 {
prev_rx8.copy_from_slice(&rx_buf[..8]);
fresh = true;
break;
}
}
Err(_) => {
rx_err += 1;
}
}
if spin_t0.elapsed().as_nanos() as u64 > timeout_ns {
break;
}
}
if !fresh {
rx_ok = false;
rx_stale += 1;
}
// jitter = deviation of fresh-data arrival interval from nominal
let now = Instant::now();
if let Some(prev) = last_fresh {
let dt = now.duration_since(prev).as_nanos() as i64 - cycle_ns as i64;
jitter_st.push(dt);
}
last_fresh = Some(now);
} else {
jitter_ns = ticker.wait_next_period();
jitter_st.push(jitter_ns);
}
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();
match ethercrab::Command::fprd(station, SM_RX_ADDR)
.receive_slice(&maindevice, SM_RX_LEN as u16)
.await
{
Ok(pdu) => {
let raw: &[u8] = &pdu;
// Strip 2-byte diagnostic word prefix (0x1A01) from plain mode
let n = raw.len().min(SM_RX_LEN);
if n > RX_PREFIX {
let copy_len = (n - RX_PREFIX).min(SM_RX_PLD_LEN);
rx_buf[..copy_len].copy_from_slice(&raw[RX_PREFIX..RX_PREFIX + copy_len]);
}
}
Err(_) => {
rx_ok = false;
rx_err += 1;
}
}
rx_rtt_st.push(t_rx0.elapsed().as_nanos() as i64);
}
if cyc <= 3 && rx_ok {
let live: u32 = rx_buf[..SM_RX_PLD_LEN.min(256)].iter().map(|&b| if b != 0 { 1 } else { 0 }).sum();
let show = SM_RX_LEN.min(64);
println!(" rx_buf[0..{}]: {} non-zero (of {}) at cycle {}", show, live, SM_RX_PLD_LEN.min(256), cyc);
print!(" rx_buf[..{}] hex: ", show);
for b in rx_buf[..show].iter() { print!("{:02X}", b); }
println!();
}
// 2. In ticker mode: prove J1900 processing by transforming TwinCAT's
// tc_seq instead of blind echo. The function f(x)=x*7+3 is a
// non-trivial mathematical transform that no hardware bridge could
// produce — it proves J1900 read, computed AND wrote the data.
// The original tc_seq is preserved at offset 272 for TwinCAT to
// verify: aIn1[0..7] == f(aOut1.tc_seq).
// In probe mode: skip TX (read-only observation).
// In rfm-bridge mode: forward the RX payload to the RFM mailbox and
// use the remote (J1900) processed REP as the TX payload, so the
// PLC-measured RTT covers the full CX->EL6695->D2000->RFM->x86 path.
if cfg.mode == Mode::RfmBridge {
if rx_ok {
let rfm_ref = rfm_ctx.as_mut().expect("rfm not opened");
let mem_ptr = rfm_ref.mem.as_ptr() as *mut u8;
let n = SM_RX_PLD_LEN.min(el6695_rt::rfm::MB_MAX_PAYLOAD);
let ts = el6695_rt::rfm::now_ns();
bridge_req_seq = bridge_req_seq.wrapping_add(1);
unsafe {
let mb = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REQ_OFF);
el6695_rt::rfm::dev_write(mb.payload.as_mut_ptr(), &rx_buf[..n]);
mb.len = n as u32;
mb.ts_ns = ts;
el6695_rt::rfm::barrier();
mb.seq = bridge_req_seq;
el6695_rt::rfm::barrier();
mb.magic = el6695_rt::rfm::MB_MAGIC;
}
// Spin-poll REP for the remote-processed payload. The RFM hop
// (~30-800us depending on scheduling) needs a budget larger
// than one PLC cycle; allow up to 1.5ms so the full-chain RTT
// is actually measured instead of timing out to local echo.
let deadline = Instant::now() + Duration::from_nanos(1_500_000);
let mut got = false;
loop {
unsafe {
let rep = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REP_OFF);
el6695_rt::rfm::barrier();
if rep.magic == el6695_rt::rfm::MB_MAGIC
&& rep.seq == bridge_req_seq
&& rep.len as usize >= SM_RX_PLD_LEN.min(el6695_rt::rfm::MB_MAX_PAYLOAD)
{
let m = SM_RX_PLD_LEN.min(el6695_rt::rfm::MB_MAX_PAYLOAD);
el6695_rt::rfm::dev_read(&mut tx_buf[..m], rep.payload.as_ptr());
// Apply f(x)=x*7+3 to the echoed frame's tc_seq
// here on the D2000 side (single transform point,
// robust): the remote just reflects the frame, the
// PLC's echo check needs [0..8] = f(rx tc_seq).
if m >= 280 {
let rx_seq = u64::from_le_bytes(rx_buf[..8].try_into().unwrap_or([0; 8]));
let processed = rx_seq.wrapping_mul(7).wrapping_add(3);
tx_buf[..8].copy_from_slice(&processed.to_le_bytes());
}
bridge_rtt_st.push(el6695_rt::rfm::now_ns().saturating_sub(rep.ts_ns) as i64);
got = true;
break;
}
}
if Instant::now() > deadline {
break;
}
core::hint::spin_loop();
}
if got {
bridge_ok += 1;
} else {
bridge_timeouts += 1;
// Fall back to local echo so downstream sees valid data
let echo_len = SM_RX_PLD_LEN.min(tx_buf.len());
tx_buf[..echo_len].copy_from_slice(&rx_buf[..echo_len]);
}
payload_checked += 1;
}
}
if cfg.mode == Mode::Ticker {
if rx_ok {
if PAYLOAD_LEN == 8 {
// 8B proof mode: per-byte transform tx[i] = rx[i] XOR 0x5A.
// Each byte is read, transformed and written back
// independently, so TwinCAT can verify all 8 bytes.
for i in 0..PAYLOAD_LEN {
tx_buf[i] = rx_buf[i] ^ 0x5A;
}
payload_checked += 1;
} else {
// Echo received data within available bounds
let echo_len = SM_RX_PLD_LEN.min(tx_buf.len());
tx_buf[..echo_len].copy_from_slice(&rx_buf[..echo_len]);
// Read TwinCAT's sequence number
let tc_seq = u64::from_le_bytes(
rx_buf[..8].try_into().unwrap_or([0; 8]),
);
// J1900 payload verification: pattern at [16..255] = (tc_seq + i) % 256
// (only present in 1KB layout)
if SM_RX_PLD_LEN >= 256 {
let mut mism: u32 = 0;
for i in 16..256 {
let expected = ((tc_seq + i as u64) % 256) as u8;
if rx_buf[i] != expected {
mism += 1;
}
}
payload_checked += 1;
payload_mismatch += mism as u64;
if PAYLOAD_LEN >= 280 {
// Write J1900 payload mismatch count for TwinCAT (offset 268)
tx_buf[268..272].copy_from_slice(&mism.to_le_bytes());
}
}
// Transform: f(x) = x * 7 + 3 (irreversible by simple subtraction)
let processed = tc_seq.wrapping_mul(7).wrapping_add(3);
tx_buf[..8].copy_from_slice(&processed.to_le_bytes());
if PAYLOAD_LEN >= 280 {
// Preserve original for TwinCAT verification (offset 272)
tx_buf[272..280].copy_from_slice(&tc_seq.to_le_bytes());
}
}
}
}
// TX (FPWR) must run for BOTH ticker and rfm-bridge modes (it was
// previously gated inside Mode::Ticker, so rfm-bridge never wrote
// SM2 and the PLC read zeros). Probe mode stays read-only.
if cfg.mode != Mode::Probe {
if PAYLOAD_LEN >= 16 {
tx_buf[8..16].copy_from_slice(&cyc.to_le_bytes());
}
if PAYLOAD_LEN >= 280 {
tx_buf[264..268].copy_from_slice(&(jitter_ns as i32).to_le_bytes());
}
if spin_mode {
// 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;
}
}
}
// Optional EL2202 scope waveform
if let Some((st2, addr)) = el2202 {
let v: u8 = if cfg.el2202_dual {
(cyc & 1) as u8 * (regs::EL2202_CH1_DATA | regs::EL2202_CH2_DATA)
} else {
(cyc & 1) as u8 * regs::EL2202_CH1_DATA
};
if ethercrab::Command::fpwr(st2, addr)
.send(&maindevice, v)
.await
.is_err()
{
el2202_err += 1;
}
}
// Optional EL2262 DC-locked reference output
if let Some(st3) = el2262 {
let v: u8 = (cyc & 1) as u8;
if ethercrab::Command::fpwr(st3, EL2262_OUT_ADDR)
.send(&maindevice, v)
.await
.is_err()
{
el2262_err += 1;
}
}
// Optional EL1252 latch readback
if let Some(st4) = el1252 {
if cyc % (cfg.el1252_every.max(1) as u64) == 0 {
match ethercrab::Command::fprd(st4, 0x09AEu16)
.receive_slice(&maindevice, 34)
.await
{
Ok(pdu) => {
let e: &[u8] = &pdu;
if e.len() >= 34 {
el1252_st = [e[0], e[1]];
let pos0 = u64::from_le_bytes(e[2..10].try_into().unwrap());
let neg0 = u64::from_le_bytes(e[10..18].try_into().unwrap());
let pos1 = u64::from_le_bytes(e[18..26].try_into().unwrap());
let neg1 = u64::from_le_bytes(e[26..34].try_into().unwrap());
if pos0 != 0 && pos0 != latch_pos0 {
if latch_pos0 != 0 {
latch_interval_st.push((pos0 - latch_pos0) as i64);
}
latch_phase_st.push((pos0 % cycle_ns) as i64);
latch_pos0 = pos0;
pos_edges += 1;
}
if neg0 != 0 && neg0 != latch_neg0 {
latch_neg0 = neg0;
neg_edges += 1;
}
if pos1 != 0 && pos1 != latch_pos1 {
if latch_pos1 != 0 {
latch1_interval_st.push((pos1 - latch_pos1) as i64);
}
latch1_phase_st.push((pos1 % cycle_ns) as i64);
latch_pos1 = pos1;
pos1_edges += 1;
}
if neg1 != 0 && neg1 != latch_neg1 {
latch_neg1 = neg1;
}
}
}
Err(_) => el1252_err += 1,
}
}
}
if rx_ok && (cfg.mode == Mode::Probe || !tx_err > 0) {
err_streak = 0;
} else {
err_streak += 1;
if err_streak > MAX_ERR_STREAK {
aborted = Some(format!(
"aborting: {} consecutive cycles with TX/RX errors (link dead?)",
err_streak
));
}
}
let tc_seq = u64::from_le_bytes(
rx_buf[..8].try_into().unwrap_or([0; 8]),
);
if tc_seq == 0 {
no_tc_seq += 1;
}
// J1900 data verification
if rx_ok && cfg.mode == Mode::Ticker {
if tc_seq == prev_tc_seq {
rx_stale += 1;
}
if prev_tc_seq != 0 && tc_seq != 0 {
let delta = (tc_seq as i64).wrapping_sub(prev_tc_seq as i64);
if delta < 0 {
rx_nonmono += 1;
}
}
prev_tc_seq = tc_seq;
// Self-consistency: verify proc == tc_seq * 7 + 3
// (always true for 1KB ticker mode; 8B mode uses XOR echo instead)
if PAYLOAD_LEN != 8 {
let computed = tc_seq.wrapping_mul(7).wrapping_add(3);
if cfg.mode == Mode::Ticker && tx_buf[..8] != computed.to_le_bytes() {
comp_err += 1; // internal computation error (should never fire)
}
}
}
exec_st.push(t0.elapsed().as_nanos() as i64);
if !cfg.quiet && last_report.elapsed() >= Duration::from_millis(1000) {
if let Some(st4) = el1252 {
if let Ok(b) = ethercrab::Command::fprd(st4, 0x1000u16)
.receive::<u8>(&maindevice)
.await
{
el1252_in = b;
}
}
let proc_val = tc_seq.wrapping_mul(7).wrapping_add(3);
let stale_pct = if cyc > 0 { (rx_stale * 100 / cyc) } else { 0 };
let mut dc_info = String::new();
if let Some((dc0, mono0)) = dc_ref {
if let Ok(dc_now) = ethercrab::Command::fprd(station, regs::DC_SYSTEM_TIME)
.receive::<u64>(&maindevice)
.await
{
let mono_now = rt::now_ns();
let drift = (dc_now.wrapping_sub(dc0)) as i64 - (mono_now - mono0) as i64;
dc_info = format!(
" dcph={} dcdr={}ns",
dc_now % cycle_ns,
drift
);
}
}
println!(
"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,
payload_mismatch,
pos_edges,
pos1_edges,
dc_info,
);
if cfg.mode == Mode::RfmBridge {
println!(
" bridge: ok={} timeouts={} rfm_hop_avg={}us",
bridge_ok,
bridge_timeouts,
if bridge_ok > 0 { bridge_rtt_st.mean() / 1000.0 } else { 0.0 },
);
}
last_report = Instant::now();
}
if aborted.is_some() {
break;
}
}
// Reap any in-flight async FPWR before reporting
if let Some(f) = pending_tx.take() {
let _ = f.await;
}
// ---- DC post-run: re-read firmware sync counters accumulated during the run ----
if cfg.dc {
println!("\n=== DC post-run sync counters (0x1C32 SM2 / 0x1C33 SM3) ===");
for idx in [0x1C32u16, 0x1C33] {
let mode = sd.sdo_read::<u16>(idx, 0x01).await.unwrap_or(0xFFFF);
let missed = sd.sdo_read::<u16>(idx, 0x0B).await.unwrap_or(0xFFFF);
let exceeded = sd.sdo_read::<u16>(idx, 0x0C).await.unwrap_or(0xFFFF);
let serr = sd.sdo_read::<u8>(idx, 0x20).await.unwrap_or(0xFF);
println!(
" 0x{:04X}: sync_mode={} sm_event_missed={} cycle_exceeded={} sync_error={}",
idx, mode, missed, exceeded, serr
);
}
}
// ---- Final report ----
println!("\n=== timing stats ===");
jitter_st.report(1000, "us");
exec_st.report(1000, "us");
rx_rtt_st.report(1000, "us");
tx_rtt_st.report(1000, "us");
if cfg.mode == Mode::RfmBridge {
bridge_rtt_st.report(1000, "us");
println!(
"rfm-bridge: ok={} timeouts={} (rep_miss_rate={:.2}%)",
bridge_ok,
bridge_timeouts,
if bridge_ok + bridge_timeouts > 0 {
bridge_timeouts as f64 * 100.0 / (bridge_ok + bridge_timeouts) as f64
} else {
0.0
}
);
}
if el1252.is_some() {
println!("=== EL1252 latch: rising-edge intervals and phase in cycle ===");
latch_interval_st.report(1000, "us");
latch_phase_st.report(1000, "us");
latch1_interval_st.report(1000, "us");
latch1_phase_st.report(1000, "us");
println!(
"el1252: ch1 pos_edges={}, ch1 neg_edges={}, ch2 pos_edges={}, read_err={}",
pos_edges, neg_edges, pos1_edges, el1252_err
);
}
println!("=== J1900 data verification ===");
println!(" tc_seq: stale={} ({}%) nonmono={} comp_err={}", rx_stale,
if cyc > 0 { rx_stale * 100 / cyc } else { 0 },
rx_nonmono, comp_err);
println!(" payload: checked={} mismatch={}", payload_checked, payload_mismatch);
println!(
"\n=== done: {} cycles in {:.1}s, tx_err={}, rx_err={}, no_tc_seq={}, el2202_err={}, el2262_err={} ===",
cyc,
start.elapsed().as_secs_f64(),
tx_err,
rx_err,
no_tc_seq,
el2202_err,
el2262_err
);
if let Some(msg) = aborted {
return Err(msg.into());
}
Ok(())
}
// =====================================================================
// RFM (PCI-5565 reflective memory) modes
// =====================================================================
/// RFM ping/echo modes: pure reflective-memory latency, no EtherCAT.
async fn rfm_mode_main(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
let rfm = el6695_rt::rfm::Rfm::open_path(cfg.rfm_dev.as_deref())?;
println!(
"rfm: {} BAR reflective mem = {} MB, node_id={}",
cfg.rfm_dev.as_deref().unwrap_or("/dev/rfm2g0(default)"),
rfm.mem.len() / (1024 * 1024),
rfm.node_id
);
let mem_ptr = rfm.mem.as_ptr() as *mut u8;
// RT-rotate this thread onto its own core (both ping and echo spin;
// sharing a core with the peer would add ~0.5ms scheduling latency).
if let Err(e) = rt::promote_current_thread(cfg.core, cfg.prio, true) {
eprintln!("RT promotion failed (continuing): {}", e);
}
// Clear both mailboxes so a stale REQ/REP from a previous run cannot
// poison the first cycle's RTT sample.
unsafe {
let zero = [0u8; 32];
el6695_rt::rfm::dev_write(mem_ptr.add(el6695_rt::rfm::MB_REQ_OFF), &zero);
el6695_rt::rfm::dev_write(mem_ptr.add(el6695_rt::rfm::MB_REP_OFF), &zero);
el6695_rt::rfm::barrier();
}
let mut rtt = OnlineStats::new("rfm_rtt", 100, 0, 5_000_000);
let mut period = OnlineStats::new("period", 100, 0, 10_000_000);
let cycle_ns = cfg.cycle_us * 1000;
let mut seq: u64 = 0;
let start = Instant::now();
let mut last_rep_seq: u64 = 0;
let mut timeouts: u64 = 0;
let mut dseq_sum: u64 = 0;
let mut last_report = Instant::now();
let mut last_cyc: u64 = 0;
match cfg.mode {
Mode::RfmPing => {
let mut next = Instant::now();
while start.elapsed().as_secs() < cfg.duration_sec || cfg.duration_sec == 0 {
if STOP.load(Ordering::Relaxed) {
break;
}
// ---- write REQ ----
let ts = el6695_rt::rfm::now_ns();
unsafe {
let mb = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REQ_OFF);
let req_seq = seq.wrapping_add(1);
let pattern = [(seq & 0xFF) as u8; 64];
el6695_rt::rfm::dev_write(mb.payload.as_mut_ptr(), &pattern);
mb.len = 64;
mb.ts_ns = ts;
el6695_rt::rfm::barrier();
mb.seq = req_seq;
el6695_rt::rfm::barrier();
mb.magic = el6695_rt::rfm::MB_MAGIC;
}
// ---- spin-wait REP ----
let deadline = Instant::now() + Duration::from_nanos(cycle_ns / 2);
let mut got = false;
loop {
unsafe {
let rep = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REP_OFF);
el6695_rt::rfm::barrier();
if rep.magic == el6695_rt::rfm::MB_MAGIC && rep.seq != last_rep_seq {
dseq_sum += seq.wrapping_add(1).saturating_sub(rep.seq);
last_rep_seq = rep.seq;
let rtt_ns = el6695_rt::rfm::now_ns().saturating_sub(rep.ts_ns);
rtt.push(rtt_ns as i64);
got = true;
break;
}
}
if Instant::now() > deadline {
break;
}
core::hint::spin_loop();
}
if !got {
timeouts += 1;
}
seq = seq.wrapping_add(1);
period.push(next.elapsed().as_nanos() as i64);
next += Duration::from_nanos(cycle_ns);
let now = Instant::now();
if next > now {
spin_sleep(next - now);
} else {
next = now;
}
if last_report.elapsed().as_secs() >= 1 {
println!(
"S tms={} cyc={} rtt_n={} timeouts={}",
start.elapsed().as_millis(),
seq,
rtt_n(&rtt),
timeouts
);
last_cyc = seq;
last_report = Instant::now();
}
}
}
Mode::RfmEcho => {
println!("rfm: echo mode (reflect REQ -> REP), ctrl-C to stop");
let mut last_req_seq: u64 = 0;
while !STOP.load(Ordering::Relaxed) {
unsafe {
let req = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REQ_OFF);
el6695_rt::rfm::barrier();
if req.magic == el6695_rt::rfm::MB_MAGIC && req.seq != last_req_seq {
let n = (req.len as usize).min(el6695_rt::rfm::MB_MAX_PAYLOAD);
let cur_seq = req.seq;
let cur_ts = req.ts_ns;
let cur_len = req.len;
let mut buf = [0u8; el6695_rt::rfm::MB_MAX_PAYLOAD];
el6695_rt::rfm::dev_read(&mut buf[..n], req.payload.as_ptr());
let rep = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REP_OFF);
el6695_rt::rfm::dev_write(rep.payload.as_mut_ptr(), &buf[..n]);
// Echo is a dumb reflector: f(x) is applied by the
// D2000 bridge (single transform point, avoids double
// transform).
rep.len = cur_len;
rep.ts_ns = cur_ts;
el6695_rt::rfm::barrier();
rep.seq = cur_seq;
el6695_rt::rfm::barrier();
rep.magic = el6695_rt::rfm::MB_MAGIC;
last_req_seq = cur_seq;
seq = seq.wrapping_add(1);
}
}
core::hint::spin_loop();
if last_report.elapsed().as_secs() >= 1 {
println!("S tms={} echoed={}", start.elapsed().as_millis(), seq);
last_report = Instant::now();
}
}
}
_ => unreachable!(),
}
println!("\n=== rfm done: {} ops in {:.1}s, timeouts={} ===", seq, start.elapsed().as_secs_f64(), timeouts);
print_stats(&rtt);
print_stats(&period);
let _ = last_cyc;
Ok(())
}
fn rtt_n(s: &OnlineStats) -> u64 {
s.len()
}
#[inline]
fn spin_sleep(d: Duration) {
let target = Instant::now() + d;
while Instant::now() < target {
if d > Duration::from_micros(200) {
std::thread::yield_now();
} else {
core::hint::spin_loop();
}
}
}
fn print_stats(s: &OnlineStats) {
s.report(1000, "us");
}