wai-quantum 0.6.0

//! The fabric's dataflow member: the gate's fixed-point arithmetic on a RISC-V
//! core of a dataflow accelerator's compute tile, reached through the device's
//! own library (loaded at run time from `WAI_QUANTUM_DATAFLOW_LIB`), which
//! presents the device as a PCI endpoint: configuration space, memory windows
//! and clocks.
//!
//! **What runs where.** The program is built in this crate ([`rv32`]): plain
//! RV32IM, with every 64-bit value a pair of words (`mul` and `mulh` for the
//! product, `sltu` for carries and borrows, the arithmetic shift from word
//! shifts). It runs on the first core of each of [`TILES`] compute tiles, each
//! over a contiguous slice of a gate's pairs: the host holds every tile's cores
//! in reset, writes the program at the core's reset address and each tile's
//! parameters and slice beside it through that tile's memory window, releases
//! the first cores, advances the clock until each program writes its done word,
//! and reads the rows back in order. It checks each row's range and keeps
//! its low 32 bits, as [`lanes::apply_words`] does.
//!
//! **Provenance.** The library this member drives is a simulator of the device,
//! so every certificate says so (`WAI_QUANTUM_DATAFLOW_SIMULATOR` names it).
//!
//! The device is named by its PCI vendor and device ids. Three mutants are
//! compiled in and must each be caught.

use crate::quantum::{Amp, Circuit, Gateset, MAX_QUBITS, ONE, QuantumError, StateVector};
use crate::quantum_fabric::lanes::{self, GateWords, Lane};
use crate::quantum_fabric::plan::{self, Config, LaneRun, Program, Step};
use crate::quantum_fabric::rv32::{self, Layout};
use crate::quantum_fabric::vector_engine::{self, Constants};
use crate::quantum_fabric::{Certificate, Certification, Corpus, FabricClass, KernelEntry, Member, MemberError, Presence, Provenance, RunFuture, RunMeter, Runner, apply_kernel};
use std::sync::{Mutex, OnceLock};

/// One probe case: [`lanes::mul_wide`] and [`lanes::fxmul_wide`] as
/// `(high, low)` words.
pub type ProbeOutput = ((u32, u32), (u32, u32));

/// Why the device is not a member, or a run has no state.
#[derive(Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum DataflowError {
    Unavailable(String),
    Circuit(QuantumError),
    TooWide { n_qubits: u8, max: u8 },
    OutOfRange { gate: u32 },
    Internal(&'static str),
    Probe { a: i32, b: i32 },
    /// The core did not finish within the clocks its program needs.
    Stalled { clocks: u64 },
}

impl std::fmt::Display for DataflowError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            DataflowError::Unavailable(why) => write!(f, "no dataflow device: {why}"),
            DataflowError::Circuit(e) => write!(f, "{e}"),
            DataflowError::TooWide { n_qubits, max } => write!(f, "{n_qubits} qubits; this member holds at most {max}"),
            DataflowError::OutOfRange { gate } => write!(f, "the result of op {gate} leaves 32 bits"),
            DataflowError::Internal(why) => write!(f, "host defect: {why}"),
            DataflowError::Probe { a, b } => write!(f, "the kernel's arithmetic differs from the CPU's at ({a}, {b})"),
            DataflowError::Stalled { clocks } => write!(f, "the core did not finish in {clocks} clocks"),
        }
    }
}

impl std::error::Error for DataflowError {}

impl From<DataflowError> for MemberError {
    fn from(e: DataflowError) -> MemberError {
        match e {
            DataflowError::TooWide { n_qubits, max } => MemberError::TooWide { n_qubits, max },
            DataflowError::OutOfRange { gate } => MemberError::OutOfRange { gate },
            DataflowError::Circuit(e) => MemberError::Circuit(e),
            // A core that ran past its clock budget: a wait past its limit.
            DataflowError::Stalled { .. } => MemberError::Timeout,
            other => MemberError::Other(other.to_string()),
        }
    }
}

/// The kernels a member runs: the conforming one, or a mutant.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum Variant {
    Conforming,
    /// The rounding constant is half what it should be.
    M1,
    /// The product is shifted by one bit too few.
    M2,
    /// Row 0's real part adds its last term instead of subtracting it.
    M3,
}

impl Variant {
    pub const MUTANTS: [Variant; 3] = [Variant::M1, Variant::M2, Variant::M3];

    /// The kernel's constants: the rounding constant, the shift, and whether
    /// row 0's last real term is added.
    pub fn constants(self) -> (u32, u32, bool) {
        match self {
            Variant::Conforming => (1 << 29, 30, false),
            Variant::M1 => (1 << 28, 30, false),
            Variant::M2 => (1 << 29, 29, false),
            Variant::M3 => (1 << 29, 30, true),
        }
    }
}

/// Where the programs put their parameters and results in the tile's memory:
/// past the programs, a batch of up to `capacity` pairs.
pub const LAYOUT: Layout = Layout { base: 0x1_0000, capacity: 8192 };

/// The rows program and the probe program of a variant, as words and listing:
/// exactly what the device runs.
pub fn programs(v: Variant) -> ((Vec<u32>, String), (Vec<u32>, String)) {
    let (rnd, sh, last_add) = v.constants();
    (rv32::rows_program(LAYOUT, rnd, sh, last_add), rv32::probe_program(LAYOUT, rnd, sh))
}

/// What runs the gate on a tile: its first RISC-V core, or the tile's vector
/// engine driven by that core.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum Engine {
    Cores,
    Vector,
}

impl Engine {
    pub fn label(self) -> &'static str {
        match self {
            Engine::Cores => "cores",
            Engine::Vector => "vector",
        }
    }
}

/// The vector engine's kernel constants for a variant.
fn vector_constants(v: Variant) -> Constants {
    let (rnd, sh, last_add) = v.constants();
    Constants { round_bit: rnd.trailing_zeros(), shift: sh, last_add }
}

/// The implementation name the id starts with.
pub const IMPLEMENTATION: &str = "rv32-lanes64";
/// The vector engine's.
pub const VECTOR_IMPLEMENTATION: &str = "sfpu-lanes64";

/// Version of the host layout: the gather of a gate's pairs into four columns,
/// the batch of [`LAYOUT`], the rows as word pairs, and the host's range check.
/// Part of the implementation id.
pub const HOST_LAYOUT: u32 = 1;

/// `"rv32-lanes64/<16 hex>"` over a variant's two listings and the layout.
pub fn implementation_id_of(v: Variant) -> String {
    let ((_, rows), (_, probe)) = programs(v);
    id_over(b"wai:quantum-fabric-dataflow\x01", IMPLEMENTATION, &[&rows, &probe])
}

pub fn implementation_id() -> String {
    implementation_id_of(Variant::Conforming)
}

/// `"sfpu-lanes64/<16 hex>"` over the vector engine's two listings (the rows
/// program with every entry zero; a gate's entries are its program's
/// immediates) and the layout.
pub fn vector_implementation_id_of(v: Variant) -> String {
    let c = vector_constants(v);
    let (_, rows) = vector_engine::rows_program(&[0; 8], c, 16);
    let (_, probe) = vector_engine::probe_program(c, 16);
    id_over(b"wai:quantum-fabric-dataflow-vector\x01", VECTOR_IMPLEMENTATION, &[&rows, &probe])
}

/// The implementation id of `engine` running `v`.
pub fn implementation_id_for(engine: Engine, v: Variant) -> String {
    match engine {
        Engine::Cores => implementation_id_of(v),
        Engine::Vector => vector_implementation_id_of(v),
    }
}

fn id_over(domain: &[u8], name: &str, texts: &[&String]) -> String {
    let mut o = domain.to_vec();
    for text in texts {
        o.extend_from_slice(&(text.len() as u64).to_be_bytes());
        o.extend_from_slice(text.as_bytes());
    }
    o.extend_from_slice(&HOST_LAYOUT.to_be_bytes());
    let h: String = blake3::hash(&o).as_bytes()[..8].iter().map(|b| format!("{b:02x}")).collect();
    format!("{name}/{h}")
}

/// The compute tiles a run spreads its pairs over, the first core of each.
pub const TILES: usize = 32;

/// The device generations this member knows: the PCI device id, the compute
/// tiles it uses, and how a memory window is set up.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct Chip {
    pub device: u16,
    /// The compute tiles' (x, y) on the device's network, as the host
    /// addresses them.
    pub tiles: &'static [(u32, u32); TILES],
    /// A window's size, and the stride of its configuration registers.
    window: u64,
    cfg_stride: u64,
    /// Whether its vector engine has the 23-bit multiplier the vector kernel
    /// is built on.
    pub vector: bool,
}

/// The PCI vendor id of the devices this member knows.
pub const VENDOR: u16 = 0x1e52;

/// Compute tiles in columns 1–4 and 6–9 of rows 1–4.
const TILES_401E: [(u32, u32); TILES] = {
    let xs = [1, 2, 3, 4, 6, 7, 8, 9];
    let mut t = [(0, 0); TILES];
    let mut i = 0;
    while i < TILES {
        t[i] = (xs[i % 8], 1 + (i / 8) as u32);
        i += 1;
    }
    t
};

/// Compute tiles in rows 2 and 3 (columns 1–7 and 10–14), then eight of row 4.
const TILES_B140: [(u32, u32); TILES] = {
    let xs = [1, 2, 3, 4, 5, 6, 7, 10, 11, 12, 13, 14];
    let mut t = [(0, 0); TILES];
    let mut i = 0;
    while i < TILES {
        t[i] = (xs[i % 12], 2 + (i / 12) as u32);
        i += 1;
    }
    t
};

pub const CHIPS: [Chip; 2] = [
    // One 64-bit configuration register per 1 MiB window.
    Chip { device: 0x401e, tiles: &TILES_401E, window: 1 << 20, cfg_stride: 8, vector: false },
    // Three 32-bit configuration registers per 2 MiB window.
    Chip { device: 0xb140, tiles: &TILES_B140, window: 1 << 21, cfg_stride: 12, vector: true },
];

/// The tile's soft-reset register: bit 11 holds the first core, 12–14 and 18
/// the others; all five set holds every core.
const SOFT_RESET: u64 = 0xffb1_21b0;
const HOLD_ALL: u32 = 0x4_7800;
const RUN_FIRST: u32 = 0x4_7000;
/// Where the window configuration registers begin in the first memory window.
const TLB_CFG: u64 = 0x1fc0_0000;

#[cfg(target_os = "linux")]
mod dl {
    use std::ffi::{c_char, c_int, c_void};
    #[link(name = "dl")]
    unsafe extern "C" {
        pub fn dlopen(filename: *const c_char, flag: c_int) -> *mut c_void;
        pub fn dlsym(handle: *mut c_void, symbol: *const c_char) -> *mut c_void;
        pub fn dlerror() -> *const c_char;
    }
    pub const RTLD_NOW: c_int = 2;
}

/// The device library's entry points.
struct Lib {
    init: unsafe extern "C" fn(),
    cfg_rd: unsafe extern "C" fn(u32, u32) -> u32,
    mem_rd: unsafe extern "C" fn(u64, *mut std::ffi::c_void, u32),
    mem_wr: unsafe extern "C" fn(u64, *const std::ffi::c_void, u32),
    clock: unsafe extern "C" fn(u32),
}

/// The device, open in this process: the library is one per process and runs
/// on the calling thread, so every access holds the lock.
struct Device {
    lib: Lib,
    chip: Chip,
    bar0: u64,
    /// The program now in every tile's memory.
    loaded: Option<Vec<u32>>,
    /// Clocks advanced so far.
    clocks: u64,
}

static DEVICE: OnceLock<Result<Mutex<Device>, String>> = OnceLock::new();

#[cfg(target_os = "linux")]
fn open_library(path: &str) -> Result<Lib, String> {
    use std::ffi::CString;
    let c = CString::new(path).map_err(|_| format!("{path:?} holds a NUL"))?;
    // SAFETY: dlopen and dlsym with valid C strings; each symbol is cast to the
    // signature the library's interface documents.
    unsafe {
        let h = dl::dlopen(c.as_ptr(), dl::RTLD_NOW);
        if h.is_null() {
            let e = dl::dlerror();
            return Err(if e.is_null() { format!("{path}: cannot load") } else { format!("{path}: {}", std::ffi::CStr::from_ptr(e).to_string_lossy()) });
        }
        let sym = |name: &str| -> Result<*mut std::ffi::c_void, String> {
            let n = CString::new(name).expect("no NUL");
            let p = dl::dlsym(h, n.as_ptr());
            if p.is_null() { Err(format!("{path} has no {name}")) } else { Ok(p) }
        };
        Ok(Lib {
            init: std::mem::transmute::<*mut std::ffi::c_void, unsafe extern "C" fn()>(sym("libttsim_init")?), // allow-name: the device library's entry point
            cfg_rd: std::mem::transmute::<*mut std::ffi::c_void, unsafe extern "C" fn(u32, u32) -> u32>(sym("libttsim_pci_config_rd32")?), // allow-name: the device library's entry point
            mem_rd: std::mem::transmute::<*mut std::ffi::c_void, unsafe extern "C" fn(u64, *mut std::ffi::c_void, u32)>(sym("libttsim_pci_mem_rd_bytes")?), // allow-name: the device library's entry point
            mem_wr: std::mem::transmute::<*mut std::ffi::c_void, unsafe extern "C" fn(u64, *const std::ffi::c_void, u32)>(sym("libttsim_pci_mem_wr_bytes")?), // allow-name: the device library's entry point
            clock: std::mem::transmute::<*mut std::ffi::c_void, unsafe extern "C" fn(u32)>(sym("libttsim_clock")?), // allow-name: the device library's entry point
        })
    }
}

#[cfg(not(target_os = "linux"))]
fn open_library(_path: &str) -> Result<Lib, String> {
    Err(format!("the device library is loaded on Linux; this is {}", std::env::consts::OS))
}

impl Device {
    fn open() -> Result<Device, String> {
        let path = std::env::var("WAI_QUANTUM_DATAFLOW_LIB").ok().filter(|s| !s.trim().is_empty()).ok_or("no device library named (WAI_QUANTUM_DATAFLOW_LIB)")?;
        let lib = open_library(path.trim())?;
        // SAFETY: the library's documented sequence: initialise once per
        // process (this runs once, under the OnceLock), then configuration
        // reads of the one endpoint at bus 0, device 0, function 0.
        let (id, lo, hi) = unsafe {
            (lib.init)();
            ((lib.cfg_rd)(0, 0), (lib.cfg_rd)(0, 0x10), (lib.cfg_rd)(0, 0x14))
        };
        let (vendor, device) = (id as u16, (id >> 16) as u16);
        if vendor != VENDOR {
            return Err(format!("the endpoint is {vendor:04x}:{device:04x}, not vendor {VENDOR:04x}"));
        }
        let chip = CHIPS.into_iter().find(|c| c.device == device).ok_or(format!("device {vendor:04x}:{device:04x} is not one this member knows"))?;
        let mut d = Device { lib, chip, bar0: (u64::from(hi) << 32) | u64::from(lo & !0xf), loaded: None, clocks: 0 };
        // Two windows per tile: its memory from 0, and its registers.
        for (j, &tile) in chip.tiles.iter().enumerate() {
            d.window(2 * j as u64, tile, 0);
            d.window(2 * j as u64 + 1, tile, SOFT_RESET & !(chip.window - 1));
            d.reg_write(j, HOLD_ALL);
        }
        Ok(d)
    }

    /// Points window `i` at `addr` (window-aligned) on `tile`.
    fn window(&mut self, i: u64, tile: (u32, u32), addr: u64) {
        let (x, y) = tile;
        let at = self.bar0 + TLB_CFG + i * self.chip.cfg_stride;
        if self.chip.window == 1 << 20 {
            let cfg = (addr >> 20) | (u64::from(x) << 16) | (u64::from(y) << 22);
            self.write(at, &(cfg as u32).to_le_bytes());
            self.write(at + 4, &((cfg >> 32) as u32).to_le_bytes());
        } else {
            let coord = u64::from(x) | (u64::from(y) << 6);
            self.write(at, &((addr >> 21) as u32).to_le_bytes());
            self.write(at + 4, &((((addr >> 53) & 0x7ff) | (coord << 11)) as u32).to_le_bytes());
            self.write(at + 8, &0u32.to_le_bytes());
        }
    }

    fn write(&mut self, paddr: u64, bytes: &[u8]) {
        // SAFETY: a write within a window the device decodes; the library
        // copies `bytes.len()` bytes from the buffer.
        unsafe { (self.lib.mem_wr)(paddr, bytes.as_ptr().cast(), bytes.len() as u32) }
    }

    fn read(&mut self, paddr: u64, out: &mut [u8]) {
        // SAFETY: as for `write`, into the buffer.
        unsafe { (self.lib.mem_rd)(paddr, out.as_mut_ptr().cast(), out.len() as u32) }
    }

    /// Writes tile `j`'s soft-reset register.
    fn reg_write(&mut self, j: usize, v: u32) {
        let at = self.bar0 + (2 * j as u64 + 1) * self.chip.window + (SOFT_RESET & (self.chip.window - 1));
        self.write(at, &v.to_le_bytes());
    }

    fn l1_write_words(&mut self, j: usize, addr: u32, words: &[u32]) {
        let bytes: Vec<u8> = words.iter().flat_map(|w| w.to_le_bytes()).collect();
        self.write(self.bar0 + 2 * j as u64 * self.chip.window + u64::from(addr), &bytes);
    }

    fn l1_read_words(&mut self, j: usize, addr: u32, n: usize) -> Vec<u32> {
        let mut bytes = vec![0u8; 4 * n];
        self.read(self.bar0 + 2 * j as u64 * self.chip.window + u64::from(addr), &mut bytes);
        bytes.as_chunks::<4>().0.iter().map(|&c| u32::from_le_bytes(c)).collect()
    }

    /// Runs `program` on the first `tiles` tiles, each with its inputs
    /// already written, until each writes 1 at `done`; a first stretch of
    /// `expected` clocks, then short ones, at most `budget` in all.
    fn run(&mut self, program: &[u32], tiles: usize, done: u32, expected: u64, budget: u64) -> Result<(), DataflowError> {
        for j in 0..TILES {
            self.reg_write(j, HOLD_ALL);
        }
        if self.loaded.as_deref() != Some(program) {
            for j in 0..TILES {
                self.l1_write_words(j, 0, program);
            }
            self.loaded = Some(program.to_vec());
        }
        for j in 0..tiles {
            self.l1_write_words(j, done, &[0]);
        }
        for j in 0..tiles {
            self.reg_write(j, RUN_FIRST);
        }
        let mut spent = 0u64;
        let mut step = expected.clamp(1, budget.min(u64::from(u32::MAX)));
        let mut waiting: Vec<usize> = (0..tiles).collect();
        loop {
            // SAFETY: advances the device's clock; nothing else is touched.
            unsafe { (self.lib.clock)(step as u32) };
            spent += step;
            self.clocks += step;
            waiting.retain(|&j| self.l1_read_words(j, done, 1)[0] != 1);
            if waiting.is_empty() {
                break;
            }
            if spent >= budget {
                for j in 0..TILES {
                    self.reg_write(j, HOLD_ALL);
                }
                return Err(DataflowError::Stalled { clocks: spent });
            }
            step = 4096.min(budget - spent).max(1);
        }
        for j in 0..tiles {
            self.reg_write(j, HOLD_ALL);
        }
        Ok(())
    }

    /// The cores' programs: each tile's pair count, then the run.
    fn run_counts(&mut self, program: &[u32], counts: &[u32], per_pair: u64) -> Result<(), DataflowError> {
        for (j, &count) in counts.iter().enumerate() {
            self.l1_write_words(j, LAYOUT.base + Layout::COUNT, &[count]);
        }
        let b = budget(counts[0], per_pair);
        self.run(program, counts.len(), LAYOUT.base + Layout::DONE, b.div_ceil(2), b)
    }
}

fn device() -> Result<&'static Mutex<Device>, String> {
    DEVICE.get_or_init(|| Device::open().map(Mutex::new)).as_ref().map_err(|e| e.clone())
}

/// What [`discover`] found.
#[derive(Debug)]
pub struct Discovery {
    pub found: Vec<DataflowMember>,
    pub refused: Vec<(String, DataflowError)>,
    pub unavailable: Option<String>,
}

/// The device, opened and probed in this process, or why there is none.
pub fn discover() -> Discovery {
    let mut d = Discovery { found: Vec::new(), refused: Vec::new(), unavailable: None };
    let chip = match device() {
        Ok(dev) => dev.lock().unwrap_or_else(|p| p.into_inner()).chip,
        Err(why) => {
            d.unavailable = Some(why);
            return d;
        }
    };
    let engines: &[Engine] = if chip.vector { &[Engine::Cores, Engine::Vector] } else { &[Engine::Cores] };
    for &engine in engines {
        match DataflowMember::open(chip, engine, Variant::Conforming) {
            Ok(m) => d.found.push(m),
            Err(e) => d.refused.push((key_for(chip, engine), e)),
        }
    }
    d
}

/// The key a runner declares for the cores: `dataflow-<vendor>-<device>`.
pub fn key_of(chip: Chip) -> String {
    format!("dataflow-{VENDOR:04x}-{:04x}", chip.device)
}

/// The key for `engine`: the vector engine's adds `-vector`.
pub fn key_for(chip: Chip, engine: Engine) -> String {
    match engine {
        Engine::Cores => key_of(chip),
        Engine::Vector => format!("{}-vector", key_of(chip)),
    }
}

/// The most qubits a run holds.
pub const MAX_DATAFLOW_QUBITS: u8 = if 20 < MAX_QUBITS { 20 } else { MAX_QUBITS };

/// A round of the vector engine: `len` pairs spread over the tiles, at most a
/// pass each, in whole pairs of iterations. An iteration covers the even or
/// the odd columns of four destination rows, so two cover the next 64 pairs
/// in order. The pairs each tile takes, and the iterations its pass runs.
fn vector_share(len: usize) -> (usize, u32) {
    let unit = 2 * vector_engine::LANES;
    let share = (len.div_ceil(TILES).div_ceil(unit).max(1) * unit).min(vector_engine::PAIRS);
    (share, (share / vector_engine::LANES) as u32)
}

/// Clocks a program needs for `count` pairs, with room to spare: the rows
/// program executes about 220 instructions a pair, the probe about 25.
fn budget(count: u32, per_pair: u64) -> u64 {
    1_000 + 2 * per_pair * u64::from(count)
}

/// A compute tile's core as a fabric member.
#[derive(Clone, Debug)]
pub struct DataflowMember {
    chip: Chip,
    engine: Engine,
    variant: Variant,
    key: String,
    id: String,
    implementation: String,
    detail: String,
    provenance: Provenance,
    max_qubits: u8,
}

impl DataflowMember {
    fn open(chip: Chip, engine: Engine, variant: Variant) -> Result<DataflowMember, DataflowError> {
        let label = std::env::var("WAI_QUANTUM_DATAFLOW_SIMULATOR").ok().filter(|s| !s.trim().is_empty()).unwrap_or_else(|| "the device library".into());
        let provenance = Provenance::Simulator(label.trim().to_string());
        let key = key_for(chip, engine);
        let mut o = b"wai:quantum-fabric-dataflow-member\x01".to_vec();
        o.extend_from_slice(key.as_bytes());
        for &(x, y) in chip.tiles {
            o.extend_from_slice(&x.to_be_bytes());
            o.extend_from_slice(&y.to_be_bytes());
        }
        o.extend_from_slice(provenance.label().as_bytes());
        let id = format!("local/dataflow-{}", blake3::hash(&o).as_bytes()[..6].iter().map(|b| format!("{b:02x}")).collect::<String>());
        let what = match engine {
            Engine::Cores => "the first RISC-V core (RV32IM)",
            Engine::Vector => "the vector engine, driven by the first RISC-V core,",
        };
        let member = DataflowMember {
            chip,
            engine,
            variant,
            detail: format!(
                "dataflow; pci {VENDOR:04x}:{:04x}, {what} of {TILES} compute tiles; {}; up to {MAX_DATAFLOW_QUBITS} qubits; unmetered",
                chip.device,
                provenance.label()
            ),
            key,
            id,
            implementation: implementation_id_for(engine, variant),
            provenance,
            max_qubits: MAX_DATAFLOW_QUBITS,
        };
        if variant == Variant::Conforming {
            let pairs = lanes::probe_pairs(1024, 0x5eed);
            let out = member.probe(&pairs)?;
            if let Some((a, b)) = lanes::first_mismatch(&pairs, &out) {
                return Err(DataflowError::Probe { a, b });
            }
        }
        Ok(member)
    }

    /// The same device running a mutant: for the mutation tests.
    #[doc(hidden)]
    pub fn with_variant(&self, v: Variant) -> DataflowMember {
        DataflowMember { variant: v, implementation: implementation_id_for(self.engine, v), ..self.clone() }
    }

    /// What runs the gate: the tiles' cores or their vector engines.
    pub fn engine(&self) -> Engine {
        self.engine
    }

    #[doc(hidden)]
    pub fn with_max_qubits(&self, max_qubits: u8) -> DataflowMember {
        DataflowMember { max_qubits: max_qubits.min(self.max_qubits), ..self.clone() }
    }

    /// The kernel this member runs: the conforming one, or a mutant.
    pub fn variant(&self) -> Variant {
        self.variant
    }

    pub fn chip(&self) -> Chip {
        self.chip
    }

    /// The clocks the device has advanced in this process.
    pub fn clocks(&self) -> u64 {
        device().map(|d| d.lock().unwrap_or_else(|p| p.into_inner()).clocks).unwrap_or(0)
    }

    pub fn id(&self) -> &str {
        &self.id
    }

    pub fn key(&self) -> &str {
        &self.key
    }

    pub fn class(&self) -> FabricClass {
        FabricClass::Dataflow
    }

    pub fn detail(&self) -> &str {
        &self.detail
    }

    pub fn implementation(&self) -> &str {
        &self.implementation
    }

    pub fn provenance(&self) -> &Provenance {
        &self.provenance
    }

    pub fn max_qubits(&self) -> u8 {
        self.max_qubits
    }

    pub fn config(&self) -> Config {
        Config::PER_GATE
    }

    fn device(&self) -> Result<std::sync::MutexGuard<'static, Device>, DataflowError> {
        Ok(device().map_err(DataflowError::Unavailable)?.lock().unwrap_or_else(|p| p.into_inner()))
    }

    /// The kernel's [`lanes::mul_wide`] and [`lanes::fxmul_wide`] of each pair.
    pub fn probe(&self, pairs: &[(i32, i32)]) -> Result<Vec<ProbeOutput>, DataflowError> {
        if self.engine == Engine::Vector {
            return self.vector_probe(pairs);
        }
        let (_, (program, _)) = programs(self.variant);
        let mut dev = self.device()?;
        let mut out = Vec::with_capacity(pairs.len());
        for round in pairs.chunks(TILES * LAYOUT.capacity as usize) {
            let slices: Vec<&[(i32, i32)]> = round.chunks(round.len().div_ceil(TILES)).collect();
            for (j, slice) in slices.iter().enumerate() {
                dev.l1_write_words(j, LAYOUT.column(0), &slice.iter().map(|p| p.0 as u32).collect::<Vec<u32>>());
                dev.l1_write_words(j, LAYOUT.column(1), &slice.iter().map(|p| p.1 as u32).collect::<Vec<u32>>());
            }
            let counts: Vec<u32> = slices.iter().map(|s| s.len() as u32).collect();
            dev.run_counts(&program, &counts, 25)?;
            for (j, slice) in slices.iter().enumerate() {
                let w = dev.l1_read_words(j, LAYOUT.results(), 4 * slice.len());
                out.extend(w.as_chunks::<4>().0.iter().map(|c| ((c[1], c[0]), (c[3], c[2]))));
            }
        }
        Ok(out)
    }

    /// The four rows of each pair, from the kernel.
    fn rows(&self, m: &[i32; 8], cols: &[Vec<i32>; 4]) -> Result<[Vec<i64>; 4], DataflowError> {
        if self.engine == Engine::Vector {
            return self.vector_rows(m, cols);
        }
        let ((program, _), _) = programs(self.variant);
        let mut dev = self.device()?;
        let n = cols[0].len();
        let mut o: [Vec<i64>; 4] = Default::default();
        for start in (0..n).step_by(TILES * LAYOUT.capacity as usize) {
            let end = (start + TILES * LAYOUT.capacity as usize).min(n);
            // Contiguous slices, one per tile, in order.
            let share = (end - start).div_ceil(TILES);
            let slices: Vec<(usize, usize)> = (start..end).step_by(share).map(|s| (s, (s + share).min(end))).collect();
            for (j, &(s, e)) in slices.iter().enumerate() {
                dev.l1_write_words(j, LAYOUT.base + Layout::ENTRIES, &m.map(|x| x as u32));
                for (k, col) in cols.iter().enumerate() {
                    dev.l1_write_words(j, LAYOUT.column(k as u32), &col[s..e].iter().map(|&v| v as u32).collect::<Vec<u32>>());
                }
            }
            let counts: Vec<u32> = slices.iter().map(|&(s, e)| (e - s) as u32).collect();
            dev.run_counts(&program, &counts, 220)?;
            for (j, &(s, e)) in slices.iter().enumerate() {
                let w = dev.l1_read_words(j, LAYOUT.results(), 8 * (e - s));
                for pair in w.as_chunks::<8>().0 {
                    for (r, row) in o.iter_mut().enumerate() {
                        row.push((i64::from(pair[2 * r + 1] as i32) << 32) | i64::from(pair[2 * r]));
                    }
                }
            }
        }
        Ok(o)
    }

    /// The vector engine's probe: up to [`vector_engine::PAIRS`] pairs a tile.
    fn vector_probe(&self, pairs: &[(i32, i32)]) -> Result<Vec<ProbeOutput>, DataflowError> {
        let n = vector_engine::PAIRS;
        let mut dev = self.device()?;
        let mut out = Vec::with_capacity(pairs.len());
        for round in pairs.chunks(TILES * n) {
            let (share, iterations) = vector_share(round.len());
            let (program, _) = vector_engine::probe_program(vector_constants(self.variant), iterations);
            let slices: Vec<&[(i32, i32)]> = round.chunks(share).collect();
            for (j, slice) in slices.iter().enumerate() {
                dev.l1_write_words(j, vector_engine::IN, &slice.iter().map(|p| p.0 as u32).collect::<Vec<u32>>());
                dev.l1_write_words(j, vector_engine::IN + 4 * n as u32, &slice.iter().map(|p| p.1 as u32).collect::<Vec<u32>>());
            }
            let clocks = vector_engine::pass_clocks(iterations);
            dev.run(&program, slices.len(), vector_engine::DONE, clocks, 20 * clocks)?;
            for (j, slice) in slices.iter().enumerate() {
                let w = dev.l1_read_words(j, vector_engine::OUT, 4 * n);
                out.extend((0..slice.len()).map(|i| ((w[n + i], w[i]), (w[3 * n + i], w[2 * n + i]))));
            }
        }
        Ok(out)
    }

    /// The vector engine's rows: up to [`vector_engine::PAIRS`] pairs a tile,
    /// the gate's entries in the program.
    fn vector_rows(&self, m: &[i32; 8], cols: &[Vec<i32>; 4]) -> Result<[Vec<i64>; 4], DataflowError> {
        let n = vector_engine::PAIRS;
        let mut dev = self.device()?;
        let len = cols[0].len();
        let mut o: [Vec<i64>; 4] = Default::default();
        for start in (0..len).step_by(TILES * n) {
            let end = (start + TILES * n).min(len);
            let (share, iterations) = vector_share(end - start);
            let (program, _) = vector_engine::rows_program(m, vector_constants(self.variant), iterations);
            let slices: Vec<(usize, usize)> = (start..end).step_by(share).map(|s| (s, (s + share).min(end))).collect();
            for (j, &(s, e)) in slices.iter().enumerate() {
                for (k, col) in cols.iter().enumerate() {
                    dev.l1_write_words(j, vector_engine::IN + (4 * n * k) as u32, &col[s..e].iter().map(|&v| v as u32).collect::<Vec<u32>>());
                }
            }
            let clocks = vector_engine::pass_clocks(iterations);
            dev.run(&program, slices.len(), vector_engine::DONE, clocks, 20 * clocks)?;
            for (j, &(s, e)) in slices.iter().enumerate() {
                let w = dev.l1_read_words(j, vector_engine::OUT, 8 * n);
                for i in 0..e - s {
                    for (r, row) in o.iter_mut().enumerate() {
                        row.push((i64::from(w[(2 * r + 1) * n + i] as i32) << 32) | i64::from(w[2 * r * n + i]));
                    }
                }
            }
        }
        Ok(o)
    }

    /// Gates in order from `state`, as [`lanes::run_words`] runs them.
    pub fn run_lanes(&self, state: &[Lane], gates: &[GateWords]) -> Result<LaneRun, DataflowError> {
        let len = state.len();
        if len < 2 || !len.is_power_of_two() {
            return Err(DataflowError::Internal("a state that is not 2^n lanes, n at least 1"));
        }
        let n = len.trailing_zeros() as u8;
        if n > self.max_qubits {
            return Err(DataflowError::TooWide { n_qubits: n, max: self.max_qubits });
        }
        let planned: Vec<(&GateWords, u32)> = gates.iter().zip(0u32..).collect();
        self.execute(state.to_vec(), &planned)
    }

    fn execute(&self, mut amps: Vec<Lane>, gates: &[(&GateWords, u32)]) -> Result<LaneRun, DataflowError> {
        let len = amps.len();
        for (w, _) in gates {
            let (tbit, ctrl) = (w.tbit as usize, w.ctrl as usize);
            if tbit == 0 || !tbit.is_power_of_two() || tbit > len / 2 || ctrl & tbit != 0 || ctrl & !(len - 1) != 0 {
                return Err(DataflowError::Internal("malformed gate words for this state"));
            }
        }
        let mut report: Option<u32> = None;
        for &(w, index) in gates {
            let m: [i32; 8] = w.m.map(|x| x as i32);
            let (tbit, ctrl) = (w.tbit as usize, w.ctrl as usize);
            let mut at = Vec::new();
            let mut cols: [Vec<i32>; 4] = Default::default();
            for p in 0..len / 2 {
                let low = p & (tbit - 1);
                let i = ((p - low) << 1) | low;
                if i & ctrl != ctrl {
                    continue;
                }
                at.push(i);
                let (a0, a1) = (amps[i], amps[i | tbit]);
                cols[0].push(a0[0]);
                cols[1].push(a0[1]);
                cols[2].push(a1[0]);
                cols[3].push(a1[1]);
            }
            if at.is_empty() {
                continue;
            }
            let [o0r, o0i, o1r, o1i] = self.rows(&m, &cols)?;
            for (k, &i) in at.iter().enumerate() {
                let rows = [o0r[k], o0i[k], o1r[k], o1i[k]];
                if rows.iter().any(|&v| i32::try_from(v).is_err()) {
                    report = Some(report.map_or(index, |r| r.min(index)));
                }
                let low = |v: i64| v as u64 as u32 as i32;
                amps[i] = [low(rows[0]), low(rows[1])];
                amps[i | tbit] = [low(rows[2]), low(rows[3])];
            }
        }
        Ok(LaneRun { amps, out_of_range: report })
    }

    pub fn run_program(&self, p: &Program) -> Result<LaneRun, DataflowError> {
        let n = p.n_qubits;
        if n == 0 {
            return Ok(LaneRun { amps: vec![[ONE as i32, 0]], out_of_range: None });
        }
        if n > self.max_qubits {
            return Err(DataflowError::TooWide { n_qubits: n, max: self.max_qubits });
        }
        if p.start >= 1usize << n {
            return Err(DataflowError::Internal("the start is outside the state"));
        }
        let mut gates = Vec::with_capacity(p.steps.len());
        for step in &p.steps {
            let Step::Gate { gate } = *step else { return Err(DataflowError::Internal("a tile step; this member runs gate steps only")) };
            let g = p.gates.get(gate as usize).ok_or(DataflowError::Internal("a step names a gate the program does not hold"))?;
            if !g.is_well_formed() {
                return Err(DataflowError::Internal("malformed gate words for this state"));
            }
            gates.push((&g.words, g.index));
        }
        let mut amps = vec![[0i32, 0]; 1usize << n];
        amps[p.start] = [ONE as i32, 0];
        self.execute(amps, &gates)
    }

    pub fn simulate_from(&self, c: &Circuit, start: usize) -> Result<StateVector, DataflowError> {
        let p = plan::compile_with(c, start, &self.config()).map_err(DataflowError::Circuit)?;
        if p.n_qubits > self.max_qubits {
            return Err(DataflowError::TooWide { n_qubits: p.n_qubits, max: self.max_qubits });
        }
        let run = self.run_program(&p)?;
        if let Some(gate) = run.out_of_range {
            return Err(DataflowError::OutOfRange { gate });
        }
        Ok(StateVector { n_qubits: p.n_qubits, amps: run.amps.iter().map(|l| Amp { re: i64::from(l[0]), im: i64::from(l[1]) }).collect() })
    }

    pub fn simulate(&self, c: &Circuit) -> Result<StateVector, DataflowError> {
        self.simulate_from(c, 0)
    }

    /// The certificate of [`apply_kernel`]`(gs)` on this member, with its
    /// provenance.
    pub fn certify(&self, gs: Gateset, corpus: Corpus) -> Result<Certificate, DataflowError> {
        let mut cert = Certification::new();
        for c in corpus.circuits(gs) {
            let reference = lanes::reference_output(&c).map_err(DataflowError::Circuit)?;
            let run = self.run_program(&plan::compile_with(&c, 0, &self.config()).map_err(DataflowError::Circuit)?)?;
            if let Some(gate) = run.out_of_range {
                return Err(DataflowError::OutOfRange { gate });
            }
            cert.add(&lanes::circuit_bytes(&c), &reference, &lanes::output(&run.amps));
        }
        Ok(cert.finish(apply_kernel(gs), &self.implementation, &self.id, &corpus.generator(gs)).with_provenance(self.provenance.clone()))
    }

    pub fn member(&self, corpus: Corpus) -> Result<Member, DataflowError> {
        let mut kernels = Vec::new();
        for gs in Gateset::ALL {
            kernels.push(KernelEntry { kernel: apply_kernel(gs).into(), implementation: self.implementation.clone(), certificate: self.certify(gs, corpus)?, energy: None });
        }
        Ok(Member { id: self.id.clone(), class: FabricClass::Dataflow, presence: Presence::Detected, kernels, detail: self.detail.clone() })
    }
}

impl Runner for DataflowMember {
    fn member_id(&self) -> String {
        self.id.clone()
    }

    fn run<'r>(&'r self, implementation: &'r str, c: &'r Circuit, start: usize, _meter: &'r mut dyn RunMeter) -> RunFuture<'r> {
        Box::pin(async move {
            if implementation != self.implementation {
                return Err(MemberError::Other(format!("member {} runs implementation {}, not {implementation:?}", self.id, self.implementation)));
            }
            Ok((self.simulate_from(c, start)?, 0, None))
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn every_variant_has_its_own_implementation_id() {
        let ids: std::collections::BTreeSet<String> = [Variant::Conforming, Variant::M1, Variant::M2, Variant::M3].into_iter().map(implementation_id_of).collect();
        assert_eq!(ids.len(), 4);
        assert!(implementation_id().starts_with("rv32-lanes64/"));
    }

    #[test]
    fn the_layout_fits_one_window_past_the_programs() {
        let ((rows, _), (probe, _)) = programs(Variant::Conforming);
        assert!(4 * rows.len().max(probe.len()) as u32 <= LAYOUT.base);
        for chip in CHIPS {
            assert!(u64::from(LAYOUT.base + LAYOUT.span()) <= chip.window);
        }
    }

    #[test]
    fn each_chip_names_distinct_tiles_whose_windows_it_has() {
        // Window registers: 156 of 1 MiB on the first, 202 of 2 MiB on the
        // second; two per tile.
        for (chip, windows) in CHIPS.into_iter().zip([156usize, 202]) {
            let set: std::collections::BTreeSet<(u32, u32)> = chip.tiles.iter().copied().collect();
            assert_eq!(set.len(), TILES);
            assert!(2 * TILES <= windows);
            assert!(2 * TILES as u64 * chip.window <= TLB_CFG, "the windows end before their configuration registers");
        }
        assert!(TILES_B140.iter().all(|&(x, y)| (2..=11).contains(&y) && !(8..=9).contains(&x) && x <= 14));
        assert!(TILES_401E.iter().all(|&(x, y)| x != 5 && x != 0 && (1..=5).contains(&y)));
    }

    #[test]
    fn a_vector_round_covers_its_pairs_in_whole_pairs_of_iterations() {
        for len in [1usize, 31, 32, 33, 1000, 1024, 1025, 16_383, 16_384] {
            let (share, it) = vector_share(len);
            assert_eq!(share, it as usize * vector_engine::LANES);
            assert!(it % 2 == 0, "{len}: an odd number of iterations leaves pairs out of order");
            assert!(share <= vector_engine::PAIRS && share * TILES >= len, "{len}: {share}");
        }
        assert_eq!(vector_share(1), (64, 2));
        assert_eq!(vector_share(TILES * vector_engine::PAIRS), (vector_engine::PAIRS, 16));
    }

    #[test]
    fn the_keys_name_the_pci_ids() {
        assert_eq!(key_of(CHIPS[0]), "dataflow-1e52-401e");
        assert_eq!(key_of(CHIPS[1]), "dataflow-1e52-b140");
    }

    #[test]
    fn with_no_library_named_the_member_is_absent_and_says_why() {
        if std::env::var_os("WAI_QUANTUM_DATAFLOW_LIB").is_none() {
            let d = discover();
            assert!(d.found.is_empty() && d.refused.is_empty());
            assert!(d.unavailable.as_deref().is_some_and(|w| w.contains("WAI_QUANTUM_DATAFLOW_LIB") || w.contains("loaded on Linux")), "{:?}", d.unavailable);
        }
    }
}