feat(rfm): PCI-5565 reflective memory via VFIO/kernel-driver, ping/echo/bridge modes

- src/rfm.rs: dual backend (rfm2g chardev /dev/rfm2g0 or vfio-pci BAR mmap),
  mailbox protocol (REQ@0x0/REP@0x1000, seq release/acquire)
- --mode rfm-ping: periodic REQ write + spin-wait REP, same-clock RTT histogram
- --mode rfm-echo: REQ->REP reflector; CX frames (>=280B) get f(x)=x*7+3
  transform so PLC echo check passes through the RFM hop
- --mode rfm-bridge: EL6695 RX -> RFM -> remote J1900 processing -> EL6695 TX
  with per-second bridge counters and rfm_hop_rtt stats (end-to-end
  CX->EL6695->D2000->RFM->x86 latency measurement)
- OnlineStats: public mean() accessor
This commit is contained in:
Tony Cao 2026-09-20 13:03:16 +08:00
parent b30e183cd8
commit e032f31392
4 changed files with 615 additions and 0 deletions

View File

@ -2,5 +2,6 @@
pub mod bridge;
pub mod regs;
pub mod rfm;
pub mod rt;
pub mod stats;

View File

@ -57,6 +57,14 @@ 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,
}
#[derive(Clone, Copy, Debug, PartialEq)]
@ -140,6 +148,9 @@ fn parse_args() -> Result<Config, String> {
"ticker" => Mode::Ticker,
"probe" => Mode::Probe,
"verify" => Mode::Verify,
"rfm-ping" => Mode::RfmPing,
"rfm-echo" => Mode::RfmEcho,
"rfm-bridge" => Mode::RfmBridge,
other => return Err(format!("unknown mode '{}'", other)),
}
}
@ -218,6 +229,9 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
Mode::Ticker => "ticker",
Mode::Probe => "probe",
Mode::Verify => "verify",
Mode::RfmPing => "rfm-ping",
Mode::RfmEcho => "rfm-echo",
Mode::RfmBridge => "rfm-bridge",
},
cfg.duration_sec,
cfg.cycle_us,
@ -226,6 +240,12 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
);
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) {
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(
@ -880,6 +900,8 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
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();
@ -930,6 +952,24 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
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 {
println!("rfm-bridge: opening PCI-5565 via VFIO...");
let r = el6695_rt::rfm::Rfm::open()?;
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);
@ -1050,6 +1090,60 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
// 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);
mb.payload[..n].copy_from_slice(&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.
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 == 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);
tx_buf[..m].copy_from_slice(&rep.payload[..m]);
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 {
@ -1303,6 +1397,14 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
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();
}
@ -1337,6 +1439,19 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
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");
@ -1369,3 +1484,166 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
}
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>> {
println!("rfm: opening PCI-5565 via VFIO...");
let rfm = el6695_rt::rfm::Rfm::open()?;
println!(
"rfm: BAR2 reflective mem = {} MB, node_id={}",
rfm.mem.len() / (1024 * 1024),
rfm.node_id
);
let mem_ptr = rfm.mem.as_ptr() as *mut u8;
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 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);
std::ptr::write_bytes(&mut mb.payload as *mut u8, (seq & 0xFF) as u8, 64);
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 {
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];
buf[..n].copy_from_slice(&req.payload[..n]);
let rep = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REP_OFF);
rep.payload[..n].copy_from_slice(&buf[..n]);
// Bridge frames (CX 800B payload): apply the same
// f(x)=x*7+3 transform on tc_seq[0..8] that the local
// ticker applies, so the PLC's echo check passes
// through the RFM hop. Small ping frames pass through
// verbatim.
if (cur_len as usize) >= 280 {
let seq_val = u64::from_le_bytes(rep.payload[..8].try_into().unwrap());
let processed = seq_val.wrapping_mul(7).wrapping_add(3);
rep.payload[..8].copy_from_slice(&processed.to_le_bytes());
}
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");
}

332
src/rfm.rs Normal file
View File

@ -0,0 +1,332 @@
//! GE/Abaco PCI-5565 (RFM2G) reflective memory access via VFIO.
//!
//! The card is bound to vfio-pci at boot (`/sbin/rfm2g-vfio-bind.sh`).
//! This module opens the VFIO container + group, obtains the device fd,
//! and mmaps BAR0 (registers) and BAR2 (64MB reflective memory).
//! Works in both SMMU (IOMMU groups) and noiommu fallback modes.
use std::fs::{File, OpenOptions};
use std::io;
use std::os::unix::io::AsRawFd;
pub const VENDOR: u16 = 0x114a;
pub const DEVICE: u16 = 0x5565;
// ---- vfio ioctls (include/uapi/linux/vfio.h) ----
const VFIO_TYPE: u8 = b';';
const VFIO_BASE: u8 = 100;
const VFIO_GET_API_VERSION: u32 = ior(VFIO_TYPE, VFIO_BASE + 0, 0);
const VFIO_GROUP_GET_DEVICE_FD: u32 = iow(VFIO_TYPE, VFIO_BASE + 6, 4);
const VFIO_DEVICE_GET_INFO: u32 = ior(VFIO_TYPE, VFIO_BASE + 7, 16);
const VFIO_DEVICE_GET_REGION_INFO: u32 = iow(VFIO_TYPE, VFIO_BASE + 8, 32);
const VFIO_API_VERSION: i32 = 0;
#[repr(C)]
#[derive(Default)]
struct VfioDeviceInfo {
argsz: u32,
flags: u32,
num_regions: u32,
num_irqs: u32,
}
#[repr(C)]
#[derive(Default, Clone, Copy)]
pub struct VfioRegionInfo {
pub argsz: u32,
pub flags: u32,
pub index: u32,
pub cap_offset: u32,
pub size: u64,
pub offset: u64,
}
const fn ioc(dir: u32, ty: u8, nr: u8, size: u32) -> u32 {
(dir << 30) | ((ty as u32) << 8) | (nr as u32) | (size << 16)
}
const fn ior(ty: u8, nr: u8, size: u32) -> u32 {
ioc(2, ty, nr, size)
}
const fn iow(ty: u8, nr: u8, size: u32) -> u32 {
ioc(1, ty, nr, size)
}
fn ioctl(fd: i32, req: u32, arg: usize) -> io::Result<i32> {
// libc::ioctl takes `Ioctl` (= c_ulong on gnu, i32 on musl) as request.
#[allow(clippy::useless_conversion)]
let r = unsafe { libc::ioctl(fd, req as _, arg) };
if r < 0 {
Err(io::Error::last_os_error())
} else {
Ok(r)
}
}
pub struct Rfm {
#[allow(dead_code)]
device_fd: File,
/// BAR0 (registers) mapping, if present.
pub regs: Option<&'static mut [u8]>,
/// BAR2 reflective memory mapping.
pub mem: &'static mut [u8],
/// Node ID read from BAR0 (byte at offset 0x34 on PCI-5565 family).
pub node_id: u8,
}
fn find_bdf() -> io::Result<String> {
for ent in std::fs::read_dir("/sys/bus/pci/devices")? {
let ent = ent?;
let p = ent.path();
let rd = |n: &str| -> Option<u32> {
std::fs::read_to_string(p.join(n))
.ok()
.and_then(|s| u32::from_str_radix(s.trim().trim_start_matches("0x"), 16).ok())
};
if rd("vendor") == Some(VENDOR as u32) && rd("device") == Some(DEVICE as u32) {
return Ok(ent.file_name().to_string_lossy().into_owned());
}
}
Err(io::Error::new(
io::ErrorKind::NotFound,
format!("PCI-5565 ({:#06x}:{:#06x}) not present", VENDOR, DEVICE),
))
}
fn iommu_group_of(bdf: &str) -> io::Result<String> {
let link = format!("/sys/bus/pci/devices/{}/iommu_group", bdf);
let t = std::fs::read_link(&link)
.map_err(|e| io::Error::new(e.kind(), format!("readlink {}: {}", link, e)))?;
Ok(t.file_name().unwrap().to_string_lossy().into_owned())
}
unsafe fn mmap_bar(fd: i32, index: u32, info: &VfioRegionInfo) -> io::Result<&'static mut [u8]> {
let map_off = (index as u64) << 40; // VFIO region mmap offset convention
let len = info.size as usize;
let p = libc::mmap(
core::ptr::null_mut(),
len,
libc::PROT_READ | libc::PROT_WRITE,
libc::MAP_SHARED,
fd,
(info.offset + map_off) as i64,
);
if p == libc::MAP_FAILED {
return Err(io::Error::last_os_error());
}
Ok(core::slice::from_raw_parts_mut(p as *mut u8, len))
}
impl Rfm {
/// Open the reflective memory card. Tries the rfm2g kernel char device
/// first (/dev/rfm2g0), falls back to VFIO.
pub fn open() -> Result<Self, Box<dyn std::error::Error>> {
if std::path::Path::new("/dev/rfm2g0").exists() {
return Self::open_chardev();
}
Self::open_vfio()
}
/// Open via the rfm2g kernel driver (/dev/rfm2g0).
///
/// The driver's mmap maps the reflective memory at pgoff == byte offset,
/// and BAR0 registers when the offset is ORed with RFM2G_BAR0_MMAP_OFFSET
/// (== 2). See rfm2g-port driver/driver_entry.c rfm2g_mmap().
fn open_chardev() -> Result<Self, Box<dyn std::error::Error>> {
const RFM2G_BAR0_MMAP_OFFSET: usize = 2;
let f = OpenOptions::new().read(true).write(true).open("/dev/rfm2g0")?;
let fd = f.as_raw_fd();
// Map first 8MB of reflective memory (mailboxes live at offset 0).
let map_len = 8 * 1024 * 1024usize;
let p = unsafe {
libc::mmap(
core::ptr::null_mut(),
map_len,
libc::PROT_READ | libc::PROT_WRITE,
libc::MAP_SHARED,
fd,
0,
)
};
if p == libc::MAP_FAILED {
return Err(io::Error::last_os_error()).map_err(|e| {
format!("rfm2g mmap failed ({}): driver loaded? card present?", e).into()
});
}
let mem = unsafe { core::slice::from_raw_parts_mut(p as *mut u8, map_len) };
// BAR0 registers for node id (best effort).
let regs: Option<&'static mut [u8]> = unsafe {
let r = libc::mmap(
core::ptr::null_mut(),
4096,
libc::PROT_READ | libc::PROT_WRITE,
libc::MAP_SHARED,
fd,
RFM2G_BAR0_MMAP_OFFSET as i64,
);
if r == libc::MAP_FAILED {
None
} else {
Some(core::slice::from_raw_parts_mut(r as *mut u8, 4096))
}
};
let node_id = regs.as_ref().map(|r| r[0x34]).unwrap_or(0xFF);
Ok(Self {
device_fd: f,
regs,
mem,
node_id,
})
}
/// Open the reflective memory card via VFIO and map BARs.
fn open_vfio() -> Result<Self, Box<dyn std::error::Error>> {
let bdf = find_bdf()?;
let group = iommu_group_of(&bdf).unwrap_or_else(|_| {
// noiommu fallback: group devices are named noiommu-N; find ours by
// matching the device link inside each noiommu group.
let mut found = String::new();
if let Ok(dir) = std::fs::read_dir("/sys/class/vfio") {
for e in dir.flatten() {
let g = e.file_name().to_string_lossy().into_owned();
if g.starts_with("noiommu-") {
let dev_link = format!("/sys/class/vfio/{}/devices/{}", g, bdf);
if std::path::Path::new(&dev_link).exists() {
found = g;
break;
}
}
}
}
found
});
if group.is_empty() {
return Err(format!("no IOMMU/noiommu group for {}", bdf).into());
}
let container = OpenOptions::new().read(true).write(true).open("/dev/vfio/vfio")?;
let group_path = format!("/dev/vfio/{}", group);
let group_fd = OpenOptions::new().read(true).write(true).open(&group_path)?;
let ver = ioctl(container.as_raw_fd(), VFIO_GET_API_VERSION, 0)?;
if ver != VFIO_API_VERSION {
return Err(format!("vfio api version {} != {}", ver, VFIO_API_VERSION).into());
}
// VFIO_GROUP_SET_CONTAINER (no IOMMU ops needed in noiommu mode)
const VFIO_GROUP_SET_CONTAINER: u32 = iow(VFIO_TYPE, VFIO_BASE + 4, 4);
let cfd = container.as_raw_fd();
ioctl(group_fd.as_raw_fd(), VFIO_GROUP_SET_CONTAINER, &cfd as *const i32 as usize)?;
let mut dev_name = bdf.clone().into_bytes();
dev_name.push(0);
let dev_fd = ioctl(
group_fd.as_raw_fd(),
VFIO_GROUP_GET_DEVICE_FD,
dev_name.as_ptr() as usize,
)?;
if dev_fd <= 0 {
return Err("VFIO_GROUP_GET_DEVICE_FD failed".into());
}
let device_fd = unsafe { File::from_raw_fd(dev_fd) };
let mut dinfo = VfioDeviceInfo {
argsz: core::mem::size_of::<VfioDeviceInfo>() as u32,
..Default::default()
};
ioctl(
device_fd.as_raw_fd(),
VFIO_DEVICE_GET_INFO,
&mut dinfo as *mut _ as usize,
)?;
let mut regs: Option<&'static mut [u8]> = None;
let mut mem: Option<&'static mut [u8]> = None;
for idx in 0..dinfo.num_regions {
let mut rinfo = VfioRegionInfo {
argsz: core::mem::size_of::<VfioRegionInfo>() as u32,
index: idx,
..Default::default()
};
if ioctl(
device_fd.as_raw_fd(),
VFIO_DEVICE_GET_REGION_INFO,
&mut rinfo as *mut _ as usize,
)
.is_err()
{
continue;
}
if rinfo.size == 0 {
continue;
}
// PCI-5565: BAR0 = registers (small), BAR2 = 64MB reflective RAM
if idx == 0 {
if let Ok(m) = unsafe { mmap_bar(device_fd.as_raw_fd(), idx, &rinfo) } {
regs = Some(m);
}
} else if idx == 2 && mem.is_none() {
mem = Some(unsafe { mmap_bar(device_fd.as_raw_fd(), idx, &rinfo)? });
}
}
let mem = mem.ok_or("BAR2 (reflective memory) not mappable")?;
let node_id = regs.as_ref().map(|r| r[0x34]).unwrap_or(0xFF);
Ok(Self {
device_fd,
regs,
mem,
node_id,
})
}
}
use std::os::unix::io::FromRawFd;
// ---- mailbox protocol over reflective memory ----
//
// Two fixed mailboxes inside BAR2. Writer fills header fields and payload,
// then bumps `seq` LAST (release). Reader polls `seq`, then reads (acquire).
// x86 and aarch64 both see coherent views of the card's onboard RAM.
pub const MB_REQ_OFF: usize = 0x0000;
pub const MB_REP_OFF: usize = 0x1000;
pub const MB_MAGIC: u32 = 0x5246_4D32; // "RFM2"
pub const MB_MAX_PAYLOAD: usize = 1024;
#[repr(C)]
pub struct Mailbox {
pub magic: u32,
pub len: u32,
pub seq: u64,
pub ts_ns: u64,
pub payload: [u8; MB_MAX_PAYLOAD],
}
impl Mailbox {
pub unsafe fn at<'a>(mem: *mut u8, off: usize) -> &'a mut Mailbox {
&mut *(mem.add(off) as *mut Mailbox)
}
}
#[inline]
pub fn barrier() {
// Full memory barrier on both x86 and aarch64.
core::sync::atomic::fence(Ordering::SeqCst);
}
use core::sync::atomic::Ordering;
#[inline]
pub fn now_ns() -> u64 {
let mut ts = libc::timespec {
tv_sec: 0,
tv_nsec: 0,
};
unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts) };
ts.tv_sec as u64 * 1_000_000_000 + ts.tv_nsec as u64
}

View File

@ -70,6 +70,10 @@ impl OnlineStats {
self.n
}
pub fn mean(&self) -> f64 {
self.mean
}
fn percentile(&self, p: f64) -> i64 {
if self.n == 0 {
return 0;