rsemu 0.0.4

//! The assembled machine: what [`realize`](mod@crate::machine::realize) produces
//! and what a run loop drives (`ROADMAP.md` §4).
//!
//! A [`Machine`] owns the four things a running machine is made of — the
//! address spaces (§4.1), the clock forest and scheduler (§4.2), the wire nets
//! (§4.3) and the device instances (§4.4) — plus the one piece of bookkeeping
//! that ties them to a snapshot: a **stable instance path** per device, which
//! is the chunk key §4.5 keys state by.
//!
//! Everything here is `no_std + alloc` and nothing names `std::sync`,
//! `std::thread` or the host clock: rate control takes a [`HostClock`]
//! injected from above the `std` line (invariant 4).
//!
//! [`HostClock`]: crate::core::sched::HostClock
//!
//! # The run loop
//!
//! ```text
//! run_quantum ─► Scheduler::run_quantum ─► Runnable::run per CPU (budgeted)
//!                                       └► events that came due
//!                    ─► Scheduler::sync_lazy_devices  (catch-up, §4.2)
//!                    ─► Instance::event per fired event
//!                    ─► Deferred::drain  after every handler
//! ```
//!
//! A quantum is bounded by the next event a **lazily advanced** device has of
//! its own, and every such device is caught up at the boundary. That is the
//! *scheduled* half of §4.2's sync-on-access; the *sampled* half fires from
//! inside the device's own `MemOps::read`, through a
//! [`LazyHandle`](crate::core::sched::LazyHandle) the realizer hands it.
//!
//! [`Machine::run_for`] is **additive** — a span taken whole and the same span
//! taken in pieces reach the same state (§11.6) — because that bound, and every
//! other instant a round may end on, is a function of virtual time and the
//! machine's state rather than of the caller's deadline. A deadline that falls
//! inside a round declines the round instead of splitting it, so a run can
//! return with up to one round's worth of time elapsed and not yet executed.
//! [`Machine::step_until`] is the one path that does split a round, for a
//! debugger, and says so.
//!
//! Which of §4.2's threading modes runs is a `core::sched` concern and the
//! loop above is the same either way:
//! [`Deterministic`](crate::core::sched::ThreadingMode::Deterministic) is the
//! mode §4.2 requires for record/replay and for the regression suite, and
//! [`Parallel`](crate::core::sched::ThreadingMode::Parallel) runs one job per
//! runnable on the task pool and joins them at the round's boundary. `accel`
//! reports itself unimplemented.
//!
//! What differs above this line is one thing: [`Machine::state_hash`] refuses
//! outside a deterministic mode, because a number a parallel run produces is a
//! sample rather than a baseline and must not be able to become a golden by
//! accident.
//!
//! Events are dispatched **after** the quantum that made them due rather than
//! from inside it, because `Scheduler::run_quantum` collects them into its
//! report. That is not a loss of precision: a quantum never runs past the next
//! deadline, so the machine is standing exactly at the event's instant when the
//! handler runs.
//!
//! # Snapshots
//!
//! [`Machine::save`] writes one chunk per device keyed by instance path, plus
//! three chunks of machine-level state: [`CLOCK_PATH`] for the oscillator
//! forest, [`SCHED_PATH`] for virtual time and the event queue, and
//! [`WIRE_PATH`] for the levels every wire source is driving. All three begin
//! with `/`, which no object name can, so they can never collide with a device.
//!
//! The scheduler chunk is there because §4.5 says the scheduler *is*
//! architectural state: the pending events, the front of virtual time and the
//! tie-break sequence counter all have to survive a load, or a restored timer
//! comes back a whole period from firing instead of the forty cycles it was
//! actually at. It is written after the clocks and read back after them too,
//! since the positions it republishes to lazily-advanced devices are derived
//! from the forest's tick counters.

use alloc::boxed::Box;
use alloc::collections::BTreeMap;
use alloc::format;
use alloc::string::{String, ToString};
use alloc::sync::Arc;
use alloc::vec::Vec;

use crate::core::clock::{ClockForest, DomainId, GlobalTime};
use crate::core::device::{Deferred, Device, DeviceClass, ResetKind};
use crate::core::error::{Error, Result};
use crate::core::record::Recorder;
use crate::core::sched::{
    Budget, Consumed, Event, EventId, EventTarget, HostClock, LazyDevice, LazyId, QuantumReport,
    Runnable, RunnableId, Scheduler, SchedulerSnapshot,
};
use crate::core::space::{AddressSpace, RequesterId};
use crate::core::state::{MachineShape, Migrations, Sink, Source, StateReader, StateWriter};
use crate::core::wire::{Level, Wire, WireId};
use crate::machine::realize::Instance;

/// The snapshot chunk holding the oscillator forest's tick counters.
pub const CLOCK_PATH: &str = "/clock";

/// The class name recorded on the [`CLOCK_PATH`] chunk.
pub const CLOCK_CLASS: &str = "machine.clock";

/// The snapshot chunk holding every wire source's level.
pub const WIRE_PATH: &str = "/wires";

/// The class name recorded on the [`WIRE_PATH`] chunk.
pub const WIRE_CLASS: &str = "machine.wires";

/// The snapshot chunk holding the scheduler: virtual time and the event queue.
pub const SCHED_PATH: &str = "/sched";

/// The class name recorded on the [`SCHED_PATH`] chunk.
pub const SCHED_CLASS: &str = "machine.sched";

/// The version of the machine-level chunks written by this build.
pub const MACHINE_STATE_VERSION: u32 = 1;

/// One address space, with the name the machine description gave it.
///
/// The space is behind an `Arc` because devices that initiate accesses need to
/// hold their own view of it (§4.4's `Initiator`). The topology is *not* frozen
/// by that: every `AddressSpace` method takes `&self`, and a retopology goes
/// through `AddressSpace::topology()`, so a BAR move or a hot-plug can still
/// remap a space this entry has already handed out.
#[derive(Debug)]
pub struct SpaceEntry {
    name: String,
    space: Arc<AddressSpace>,
}

impl SpaceEntry {
    /// The space's name, as the `space` statement spelled it.
    pub fn name(&self) -> &str {
        &self.name
    }

    /// The space itself.
    pub fn space(&self) -> &Arc<AddressSpace> {
        &self.space
    }
}

/// One device instance and everything the machine knows about it.
#[derive(Debug)]
pub struct DeviceEntry {
    pub(crate) path: String,
    pub(crate) class: &'static DeviceClass,
    pub(crate) device: Arc<dyn Device>,
    pub(crate) instance: Option<Arc<dyn Instance>>,
    pub(crate) domain: Option<DomainId>,
    pub(crate) space: Option<usize>,
    pub(crate) requester: RequesterId,
    pub(crate) runnable: Option<RunnableId>,
    pub(crate) lazy: Option<LazyId>,
}

impl DeviceEntry {
    /// The instance path — the snapshot chunk key (§4.5), stable for the life
    /// of the machine.
    pub fn path(&self) -> &str {
        &self.path
    }

    /// The class this instance was built from.
    pub fn class(&self) -> &'static DeviceClass {
        self.class
    }

    /// The device.
    pub fn device(&self) -> &Arc<dyn Device> {
        &self.device
    }

    /// The device's machine-layer view, when its class is bound (see
    /// [`Bindings`](crate::machine::realize::Bindings)).
    pub fn instance(&self) -> Option<&Arc<dyn Instance>> {
        self.instance.as_ref()
    }

    /// The clock domain it runs in, if it declared one.
    pub fn domain(&self) -> Option<DomainId> {
        self.domain
    }

    /// The address space it declared, as an index into [`Machine::spaces`].
    pub fn space_index(&self) -> Option<usize> {
        self.space
    }

    /// Its requester id, as it appears in `MemAttrs` for accesses it initiates.
    pub fn requester(&self) -> RequesterId {
        self.requester
    }

    /// Its scheduler handle, if it takes execution budgets.
    pub fn runnable(&self) -> Option<RunnableId> {
        self.runnable
    }

    /// Its catch-up handle, if it declared itself lazily advanced (§4.2).
    pub fn lazy(&self) -> Option<LazyId> {
        self.lazy
    }
}

/// One wire net: a set of pins that are the same piece of copper.
#[derive(Debug)]
pub struct Net {
    pub(crate) wire: Arc<Wire>,
    pub(crate) sources: Vec<PinRef>,
}

impl Net {
    /// The net itself.
    pub fn wire(&self) -> &Arc<Wire> {
        &self.wire
    }

    /// The pins driving it, in the order their ids were allocated.
    pub fn sources(&self) -> &[PinRef] {
        &self.sources
    }
}

/// One end of a wire: a device and one of its pins.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PinRef {
    /// Index into [`Machine::devices`].
    pub device: usize,
    /// The pin's name, as the device knows it.
    pub port: String,
    /// The id this pin drives the net with, for a source pin.
    pub id: WireId,
}

/// Whether a run may split the scheduling round its deadline lands in.
///
/// [`Stepping::Whole`] is what a run loop wants and what keeps
/// [`Machine::run_for`] additive; [`Stepping::Fragment`] is the debugger's.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Stepping {
    /// Stop at the deadline, having run only whole rounds.
    Whole,
    /// Run whatever fits before the deadline.
    Fragment,
}

/// The `Runnable` the scheduler sees, wrapping the `Instance` the machine owns.
///
/// The shim exists because the two halves disagree about mutability: §4.6's
/// `Cpu::run` takes `&self` (a device is shared — it is `Send + Sync` with
/// interior mutability), while [`Runnable::run`] takes `&mut self` and the
/// scheduler takes ownership of the box. Forwarding through an `Arc` is the
/// only way to satisfy both without the machine giving up ownership of its own
/// device.
pub(crate) struct RunAdapter {
    inner: Arc<dyn Instance>,
}

impl RunAdapter {
    /// Wrap `inner` so the scheduler can hand it budgets.
    pub(crate) fn new(inner: Arc<dyn Instance>) -> RunAdapter {
        RunAdapter { inner }
    }
}

impl Runnable for RunAdapter {
    fn run(&mut self, budget: Budget) -> Consumed {
        self.inner.run(budget)
    }
}

/// The [`LazyDevice`] the scheduler sees, wrapping the device the machine owns.
///
/// The same shim as [`RunAdapter`] and for the same reason: a `Device` declares
/// itself lazily advanced through `&self` methods, because a device is shared
/// and holds its state behind interior mutability, while
/// [`LazyDevice::advance_to`] takes `&mut self`. Forwarding through an `Arc` is
/// what satisfies both.
///
/// The `&mut` is not wasted. The scheduler takes the box *out* of its slot for
/// the duration of the call (`core::sched`), so the exclusive borrow is what
/// makes a device that re-enters its own catch-up get
/// [`SchedError::LazyDeviceBusy`](crate::core::sched::SchedError::LazyDeviceBusy)
/// rather than a deadlock.
pub(crate) struct LazyAdapter {
    inner: Arc<dyn Device>,
}

impl LazyAdapter {
    /// Wrap `inner` so the scheduler can catch it up.
    pub(crate) fn new(inner: Arc<dyn Device>) -> LazyAdapter {
        LazyAdapter { inner }
    }
}

impl LazyDevice for LazyAdapter {
    fn current_tick(&self) -> u64 {
        self.inner.current_tick()
    }

    fn advance_to(&mut self, tick: u64) {
        // `&self` on the device side: `&mut self` here is the scheduler's
        // exclusivity, not the device's.
        self.inner.advance_to(tick);
    }

    fn next_event_tick(&self) -> Option<u64> {
        self.inner.next_event_tick()
    }

    fn sampled_every_cycle(&self) -> bool {
        self.inner.sampled_every_cycle()
    }
}

/// A realized machine: spaces, clocks, wires and devices, ready to run.
///
/// Built by [`realize`](crate::machine::realize::realize). Nothing observable
/// happens before that call returns, so a description that fails half way
/// leaves no half-wired machine behind (§4.4).
#[derive(Debug)]
pub struct Machine {
    name: String,
    spaces: Vec<SpaceEntry>,
    sched: Scheduler,
    devices: Vec<DeviceEntry>,
    by_path: BTreeMap<String, usize>,
    nets: Vec<Net>,
    sweep: Vec<PinRef>,
    shape: MachineShape,
    deferred: Deferred,
    /// The record/replay seam, if one is attached (§4.5).
    ///
    /// `None` is the ordinary case and costs one `Option` test per scheduling
    /// round, which is nothing against a round. It is deliberately *not* part
    /// of [`MachineParts`]: a recorder is a property of the run, like the
    /// threading mode, and attaching one has to be able to fail — see
    /// [`Machine::set_recorder`].
    recorder: Option<Arc<Recorder>>,
}

/// The parts a realizer hands to [`Machine::assemble`].
///
/// A struct rather than eight positional arguments: every one of them is a
/// `Vec` or a name, and swapping two at a call site would compile.
#[derive(Debug)]
pub(crate) struct MachineParts {
    pub(crate) name: String,
    pub(crate) spaces: Vec<(String, Arc<AddressSpace>)>,
    pub(crate) sched: Scheduler,
    pub(crate) devices: Vec<DeviceEntry>,
    pub(crate) nets: Vec<Net>,
    pub(crate) sweep: Vec<PinRef>,
    pub(crate) shape: MachineShape,
    pub(crate) deferred: Deferred,
}

impl Machine {
    /// Assemble a machine from parts. Called by the realizer, and by nothing
    /// else — every field has an invariant the realizer establishes.
    pub(crate) fn assemble(parts: MachineParts) -> Machine {
        let by_path = parts
            .devices
            .iter()
            .enumerate()
            .map(|(i, d)| (d.path.clone(), i))
            .collect();
        Machine {
            name: parts.name,
            spaces: parts
                .spaces
                .into_iter()
                .map(|(name, space)| SpaceEntry { name, space })
                .collect(),
            sched: parts.sched,
            devices: parts.devices,
            by_path,
            nets: parts.nets,
            sweep: parts.sweep,
            shape: parts.shape,
            deferred: parts.deferred,
            recorder: None,
        }
    }

    /// The machine's name, as `machine "nes"` wrote it.
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Every address space, in declaration order.
    pub fn spaces(&self) -> &[SpaceEntry] {
        &self.spaces
    }

    /// One address space by name.
    pub fn space(&self, name: &str) -> Option<&Arc<AddressSpace>> {
        self.spaces
            .iter()
            .find(|s| s.name == name)
            .map(SpaceEntry::space)
    }

    /// Every device instance, in declaration order — which is also reset order.
    pub fn devices(&self) -> &[DeviceEntry] {
        &self.devices
    }

    /// One device by instance path.
    pub fn device(&self, path: &str) -> Option<&DeviceEntry> {
        self.by_path.get(path).and_then(|i| self.devices.get(*i))
    }

    /// The index of the device at `path`, for scheduling events against it.
    pub fn device_index(&self, path: &str) -> Option<usize> {
        self.by_path.get(path).copied()
    }

    /// Every wire net, in the order the realizer built them.
    pub fn nets(&self) -> &[Net] {
        &self.nets
    }

    /// The scheduler, which owns virtual time and the clock forest.
    pub fn scheduler(&self) -> &Scheduler {
        &self.sched
    }

    /// The scheduler, mutably — for posting events and changing rate control.
    pub fn scheduler_mut(&mut self) -> &mut Scheduler {
        &mut self.sched
    }

    /// The oscillator forest (§4.2).
    pub fn clocks(&self) -> &ClockForest {
        self.sched.forest()
    }

    /// The current virtual instant.
    pub fn now(&self) -> GlobalTime {
        self.sched.now()
    }

    /// The machine's structural fingerprint, which a snapshot is checked
    /// against (§4.5).
    pub fn shape(&self) -> &MachineShape {
        &self.shape
    }

    /// Inject the host's monotonic clock, for rate control.
    ///
    /// Nothing below `host/` may read a wall clock (invariant 4), so the
    /// machine is handed one rather than reaching for it.
    pub fn set_host_clock(&mut self, clock: Box<dyn HostClock>) {
        self.sched.set_host_clock(clock);
    }

    // -----------------------------------------------------------------
    // reset and the realize sweep
    // -----------------------------------------------------------------

    /// Reset every device, then re-announce every wire source (§4.3).
    ///
    /// Devices are reset in **declaration order** — the order the machine file
    /// names them — because a reset order that depends on a hash or on the
    /// wiring is a reset order that changes between runs, and §0 does not allow
    /// that. A device that must be reset after another says so by being
    /// declared after it.
    ///
    /// The deferred queue is drained after each device, so an action a reset
    /// handler pushes runs before the next device is touched, in the order it
    /// was pushed.
    pub fn reset(&mut self, kind: ResetKind) {
        for i in 0..self.devices.len() {
            let device = Arc::clone(&self.devices[i].device);
            device.reset(kind);
            self.deferred.drain();
        }
        self.sweep();
    }

    /// The realize sweep: walk wire sources in topological order and announce
    /// the level each drives (§4.3).
    ///
    /// Without it a freshly realized — or freshly restored — machine is
    /// inconsistent: an undriven wire sits low, which contradicts an inverter
    /// whose output idles high, and the interrupt line comes up wrong on some
    /// machines and only on some paths. The order is
    /// [`realize_order`](crate::machine::validate::realize_order)'s, so a
    /// source announces before anything that forwards its level.
    pub fn sweep(&mut self) {
        for pin in &self.sweep {
            if let Some(instance) = self.devices[pin.device].instance.as_ref() {
                instance.announce(&pin.port);
            }
        }
        self.deferred.drain();
    }

    // -----------------------------------------------------------------
    // running
    // -----------------------------------------------------------------

    /// Run one scheduler quantum and dispatch whatever came due.
    ///
    /// # Errors
    ///
    /// Whatever the scheduler refuses — an overrun budget, an unimplemented
    /// threading mode — or an event addressed to a device that does not exist.
    pub fn run_quantum(&mut self) -> Result<QuantumReport> {
        self.pump_inputs()?;
        let report = self.sched.run_quantum_until(GlobalTime::MAX)?;
        self.sched.sync_lazy_devices()?;
        self.dispatch(&report)?;
        Ok(report)
    }

    /// Run until virtual time reaches `deadline`.
    ///
    /// The loop is here rather than in `Scheduler::run_until` because that one
    /// discards the per-quantum report, and the report is where fired events
    /// are: driving it from above is what keeps them from being dropped.
    ///
    /// # Errors
    ///
    /// As [`Machine::run_quantum`], plus a machine whose configuration cannot
    /// advance virtual time at all.
    pub fn run_until(&mut self, deadline: GlobalTime) -> Result<()> {
        self.advance_to(deadline, Stepping::Whole)
    }

    /// Advance to `deadline`, cutting the round it lands in.
    ///
    /// **For a debugger, and not additive.** [`Machine::run_until`] declines a
    /// round the deadline falls inside, because splitting one is the scheduling
    /// boundary that made `run_for` non-additive (§11.6). A debugger stepping a
    /// cycle at a time cannot afford that: waiting for the round to end would
    /// step over every breakpoint between here and its boundary. So this asks
    /// for the fragment, explicitly, at the cost of the property `run_until`
    /// exists to keep.
    ///
    /// # Errors
    ///
    /// As [`Machine::run_quantum`].
    pub fn step_until(&mut self, deadline: GlobalTime) -> Result<()> {
        self.advance_to(deadline, Stepping::Fragment)
    }

    fn advance_to(&mut self, deadline: GlobalTime, stepping: Stepping) -> Result<()> {
        while self.sched.now() < deadline {
            let before = self.sched.now();
            self.pump_inputs()?;
            let report = match stepping {
                Stepping::Whole => self.sched.run_quantum_until(deadline)?,
                Stepping::Fragment => self.sched.step_quantum_until(deadline)?,
            };
            // Before the events are dispatched: a handler that reads a lazily
            // advanced device must see it standing on the instant that fired,
            // not on the one the previous quantum ended at.
            self.sched.sync_lazy_devices()?;
            self.dispatch(&report)?;
            if self.sched.now() <= before {
                // A quantum ends either at its natural boundary or, when the
                // deadline falls before that, at the deadline itself — and the
                // deadline is above `now` or this loop would not have run. So
                // this can only mean a zero quantum, whose grid has every point
                // on top of every other. Reporting it is the only honest
                // option: spinning would hang, and jumping to the deadline
                // through `Scheduler::run_until` would fire events into a
                // report nobody reads.
                return Err(Error::Config {
                    at: self.name.clone(),
                    message: "virtual time did not advance: the scheduler quantum is zero"
                        .to_string(),
                });
            }
        }
        Ok(())
    }

    /// Run for `span` of virtual time from wherever the machine is now.
    ///
    /// Additive: running for a span and running for the same span in pieces
    /// reach the same state (§11.6). What that costs is that the run stops on
    /// the machine's own scheduling boundaries, so it can return with up to one
    /// round of virtual time elapsed but not yet executed — the next call
    /// executes it, and nothing is lost.
    ///
    /// # Errors
    ///
    /// As [`Machine::run_quantum`].
    pub fn run_for(&mut self, span: GlobalTime) -> Result<()> {
        let deadline = self.sched.now().saturating_add(span);
        self.run_until(deadline)
    }

    /// Post an event for the device at `path`, `ticks` of its own clock domain
    /// from now.
    ///
    /// `token` is handed back to the device untouched: it is a timer index, a
    /// channel number, whatever the device put there.
    ///
    /// # Errors
    ///
    /// If no device is at `path`, if it has no clock domain, or if the clock
    /// conversion overflows.
    pub fn schedule_after_ticks(&mut self, path: &str, ticks: u64, token: u64) -> Result<EventId> {
        let index = self.device_index(path).ok_or_else(|| Error::Config {
            at: path.to_string(),
            message: "no device at this instance path".to_string(),
        })?;
        let domain = self.devices[index].domain.ok_or_else(|| Error::Config {
            at: path.to_string(),
            message: "cannot post an event for a device with no clock domain".to_string(),
        })?;
        let target = EventTarget(u32::try_from(index).unwrap_or(u32::MAX));
        Ok(self
            .sched
            .schedule_after_ticks(domain, ticks, target, token)?)
    }

    /// Deliver every event in `report` to its device, draining the deferred
    /// queue after each handler.
    ///
    /// The drain is per handler, not per quantum: an action a device defers is
    /// meant to run *after that handler returns* and before anything else
    /// observes the machine (§4.7's re-entrancy contract). Batching them to the
    /// end of the quantum would reorder them against the next event.
    fn dispatch(&mut self, report: &QuantumReport) -> Result<()> {
        for event in &report.fired {
            let index = event.target.0 as usize;
            let Some(instance) = self
                .devices
                .get(index)
                .map(|d| d.instance.clone())
                .ok_or_else(|| Error::Config {
                    at: self.name.clone(),
                    message: format!(
                        "event {} is addressed to device {index}, which does not exist",
                        event.id.seq()
                    ),
                })?
            else {
                // A device with no machine-layer view cannot have posted an
                // event, so this is a stale target rather than a lost handler.
                continue;
            };
            instance.event(event.token, &mut self.deferred);
            self.deferred.drain();
        }
        Ok(())
    }

    // -----------------------------------------------------------------
    // snapshots
    // -----------------------------------------------------------------

    /// Serialize the whole machine: one chunk per device, keyed by instance
    /// path, plus the clock forest, the scheduler and the wire levels (§4.5).
    ///
    /// # Errors
    ///
    /// Whatever a device's `save` reports, or a duplicate instance path.
    pub fn save(&self) -> Result<Vec<u8>> {
        let mut w = StateWriter::new(self.shape.clone());
        for entry in &self.devices {
            let mut chunk = w.chunk(&entry.path, entry.class.name, entry.class.version)?;
            entry.device.save(&mut chunk)?;
        }
        {
            let mut chunk = w.chunk(CLOCK_PATH, CLOCK_CLASS, MACHINE_STATE_VERSION)?;
            save_clocks(self.sched.forest(), &mut chunk)?;
        }
        {
            let mut chunk = w.chunk(SCHED_PATH, SCHED_CLASS, MACHINE_STATE_VERSION)?;
            save_sched(&self.sched, &mut chunk)?;
        }
        {
            let mut chunk = w.chunk(WIRE_PATH, WIRE_CLASS, MACHINE_STATE_VERSION)?;
            save_wires(&self.nets, &mut chunk)?;
        }
        w.to_vec()
    }

    /// Restore what [`Machine::save`] wrote, with no class migrations.
    ///
    /// # Errors
    ///
    /// As [`Machine::load_with`].
    pub fn load(&mut self, bytes: &[u8]) -> Result<()> {
        self.load_with(bytes, &Migrations::new())
    }

    /// Restore a snapshot, migrating device chunks through `migrations`.
    ///
    /// The machine's shape is checked first, so a snapshot taken from a
    /// differently-shaped machine fails with a diff naming what moved rather
    /// than by loading nonsense into the wrong device (§4.5).
    ///
    /// The realize sweep runs afterwards: a restored machine is as inconsistent
    /// as a fresh one until every gate drives what its inputs imply.
    ///
    /// # Errors
    ///
    /// A shape mismatch, a missing or mis-classed chunk, a migration hole, or
    /// whatever a device's `load` reports.
    pub fn load_with(&mut self, bytes: &[u8], migrations: &Migrations) -> Result<()> {
        let reader = StateReader::new(bytes)?;
        reader.check_shape(&self.shape)?;
        for entry in &self.devices {
            let chunk = reader.load(
                &entry.path,
                entry.class.name,
                entry.class.version,
                migrations,
            )?;
            let mut r = chunk.reader();
            entry.device.load(&mut r)?;
        }
        let clocks = reader.load(CLOCK_PATH, CLOCK_CLASS, MACHINE_STATE_VERSION, migrations)?;
        load_clocks(self.sched.forest_mut(), &mut clocks.reader())?;
        // After the clocks: the scheduler's restore republishes every lazily
        // advanced device's domain position, which is only right once the tick
        // counters those positions come from are back.
        let sched = reader.load(SCHED_PATH, SCHED_CLASS, MACHINE_STATE_VERSION, migrations)?;
        load_sched(&mut self.sched, &mut sched.reader())?;
        let wires = reader.load(WIRE_PATH, WIRE_CLASS, MACHINE_STATE_VERSION, migrations)?;
        load_wires(&self.nets, &mut wires.reader())?;
        self.deferred.drain();
        self.sweep();
        // A load moves virtual time, so the input seam has to move with it or
        // the next round delivers against an instant the machine has left. This
        // is what makes `Machine::load` sound on its own rather than only
        // inside [`Timeline::rewind_to`](crate::machine::Timeline::rewind_to):
        // a debugger restoring a snapshot gets the same treatment.
        if let Some(recorder) = &self.recorder {
            recorder.rewind_to(self.sched.now());
        }
        Ok(())
    }

    /// A hash of the machine's serialized state.
    ///
    /// The regression method of §0 in one call: run deterministically for N
    /// virtual units and compare this number. It is a hash of [`Machine::save`]
    /// output, which `core::state` guarantees is byte-identical for identical
    /// state, so equal hashes mean equal state and not merely equal-looking
    /// state.
    ///
    /// # Only in a deterministic machine, and that is structural
    ///
    /// This **refuses** on a machine whose threading mode is not
    /// [`ThreadingMode::Deterministic`]. `ROADMAP.md` §0 makes reproducibility
    /// a property of the mode rather than of the thread count, and §4.2 calls
    /// [`ThreadingMode::Parallel`] non-deterministic in as many words — so a
    /// number taken from a parallel run is not a regression baseline, it is a
    /// sample. Refusing here is what makes that structural instead of a comment
    /// somebody has to have read: a conformance suite, a frame-hash golden or a
    /// replay trace cannot be blessed against a parallel run by accident,
    /// because the call that would produce the number returns an error.
    ///
    /// [`Machine::nondeterministic_state_hash`] is for the caller that wants
    /// the number anyway — a snapshot round-trip inside one run, where both
    /// sides come from the same execution and no golden is being written. Its
    /// name is the documentation.
    ///
    /// # Errors
    ///
    /// [`Error::Config`] if the machine is not in a deterministic threading
    /// mode; otherwise as [`Machine::save`].
    ///
    /// [`ThreadingMode`]: crate::core::sched::ThreadingMode
    /// [`ThreadingMode::Deterministic`]: crate::core::sched::ThreadingMode::Deterministic
    /// [`ThreadingMode::Parallel`]: crate::core::sched::ThreadingMode::Parallel
    pub fn state_hash(&self) -> Result<u64> {
        let mode = self.sched.config().mode;
        if !mode.is_deterministic() {
            return Err(Error::Config {
                at: self.name.clone(),
                message: format!(
                    "a state hash from `{mode}` threading is not reproducible and must not \
                     become a golden (ROADMAP.md 4.2); run the machine in `deterministic` \
                     threading, or say `nondeterministic_state_hash` and mean it"
                ),
            });
        }
        Ok(fnv1a(&self.save()?))
    }

    /// A hash of the machine's serialized state, whatever the threading mode.
    ///
    /// The escape hatch [`Machine::state_hash`] describes. Legitimate when both
    /// sides of the comparison come from **one run** — a snapshot taken and
    /// restored inside the same execution, a device's state before and after a
    /// reset. Never legitimate as a value checked into a test, a ledger or a
    /// golden file, because a parallel run does not produce the same one twice.
    ///
    /// # Errors
    ///
    /// As [`Machine::save`].
    pub fn nondeterministic_state_hash(&self) -> Result<u64> {
        Ok(fnv1a(&self.save()?))
    }

    // -----------------------------------------------------------------
    // record / replay (§4.5)
    // -----------------------------------------------------------------

    /// Attach the record/replay seam.
    ///
    /// From here on, the top of every scheduling round drains the recorder:
    /// in [`Mode::Record`] whatever the host has posted since the last round is
    /// stamped with the round's own instant, logged and delivered; in
    /// [`Mode::Replay`] the logged events due at that instant are delivered and
    /// the host is ignored. [`core::record`](crate::core::record) argues why
    /// delivery has to happen there and nowhere else.
    ///
    /// The machine's [`MachineShape`] is written onto the log, so a recording
    /// taken here and replayed into a different board fails with a diff.
    ///
    /// # Only in a deterministic machine, and that is structural
    ///
    /// This **refuses** on a machine whose threading mode is not
    /// [`ThreadingMode::Deterministic`], exactly as [`Machine::state_hash`]
    /// does and for the same reason. `ROADMAP.md` §4.2 makes
    /// [`ThreadingMode::Deterministic`] a requirement of record/replay in as
    /// many words, and the honest reason is narrower than "parallel is
    /// non-deterministic": `Scheduler::run_quantum` *does* join every job
    /// before a round returns, so the round boundaries this seam timestamps
    /// against are reproducible under `parallel` too. What is not reproducible
    /// is what happens **inside** a round — two CPU threads interleaving their
    /// accesses to shared memory in an order the host picks, and reporting back
    /// tick counts that depend on what they read. No input log can recover
    /// that, so a recording taken from a parallel run would replay into a
    /// different machine while looking entirely valid. Refusing here is what
    /// makes that impossible rather than merely documented.
    ///
    /// # Errors
    ///
    /// [`Error::Config`] if the machine is not in a deterministic threading
    /// mode.
    ///
    /// [`Mode::Record`]: crate::core::record::Mode::Record
    /// [`Mode::Replay`]: crate::core::record::Mode::Replay
    /// [`ThreadingMode`]: crate::core::sched::ThreadingMode
    /// [`ThreadingMode::Deterministic`]: crate::core::sched::ThreadingMode::Deterministic
    pub fn set_recorder(&mut self, recorder: Arc<Recorder>) -> Result<()> {
        let mode = self.sched.config().mode;
        if !mode.is_deterministic() {
            return Err(Error::Config {
                at: self.name.clone(),
                message: format!(
                    "a recording of a `{mode}` run cannot be replayed: the round boundaries \
                     are reproducible but what happens inside a round is not, so no input log \
                     can restore it (ROADMAP.md 4.2). Run the machine in `deterministic` \
                     threading to record it"
                ),
            });
        }
        recorder.set_shape(self.shape.clone());
        self.recorder = Some(recorder);
        Ok(())
    }

    /// Detach the seam, returning whatever was attached.
    pub fn take_recorder(&mut self) -> Option<Arc<Recorder>> {
        self.recorder.take()
    }

    /// The attached seam, if there is one.
    #[inline]
    pub fn recorder(&self) -> Option<&Arc<Recorder>> {
        self.recorder.as_ref()
    }

    /// Deliver the input due at this instant.
    ///
    /// Called at the top of every scheduling round, which is the only place a
    /// non-deterministic input may enter a machine. Standing exactly on a round
    /// boundary is what makes the instant a function of the guest's timeline
    /// rather than of the host thread that posted.
    fn pump_inputs(&mut self) -> Result<()> {
        if let Some(recorder) = &self.recorder {
            recorder.deliver(self.sched.now())?;
        }
        Ok(())
    }

    /// The threading mode this machine runs in (§4.2).
    #[inline]
    pub fn threading_mode(&self) -> crate::core::sched::ThreadingMode {
        self.sched.config().mode
    }

    /// The stop-the-world protocol (§4.7).
    ///
    /// Clone it to whatever may need the machine quiescent — a host thread
    /// wanting a snapshot, a device that remaps memory from inside its own
    /// write path. See [`SafePoint`](crate::core::sched::SafePoint).
    #[inline]
    pub fn safe_point(&self) -> crate::core::sched::SafePoint {
        self.sched.safe_point()
    }

    /// Stop the world and hold it stopped until the guard is dropped.
    ///
    /// Every runnable that honours its exit flag declines to start another
    /// block, and the task pool is quiesced. What comes back is a machine
    /// nobody is executing, which is what a snapshot, a retopology or a reset
    /// needs (§4.7). Under [`ThreadingMode::Deterministic`] it is nearly free
    /// and still correct: there was never anybody else to stop.
    ///
    /// [`ThreadingMode::Deterministic`]: crate::core::sched::ThreadingMode::Deterministic
    #[must_use = "the world runs again when the guard is dropped"]
    pub fn stop_the_world(&self) -> crate::core::sched::StopGuard {
        self.sched.stop_the_world()
    }
}

/// FNV-1a over the snapshot bytes.
///
/// Not a cryptographic hash and not meant to be: `purecrypto`'s BLAKE3 is the
/// integrity seam (§4.5), and this is a test and regression comparison that has
/// to work in a dependency-free `no_std` build.
fn fnv1a(bytes: &[u8]) -> u64 {
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    for b in bytes {
        h ^= u64::from(*b);
        h = h.wrapping_mul(0x0000_0100_0000_01b3);
    }
    h
}

/// Write every oscillator's unit position and every domain's tick counter.
///
/// The tick counters are the authoritative architectural state (§4.2); the
/// global timeline is derived from them and is recomputed on restore.
///
/// Both sequences come from the forest's own enumeration, in creation order, so
/// the writer and the reader agree without either of them having to have kept a
/// list of handles.
fn save_clocks(forest: &ClockForest, sink: &mut impl Sink) -> Result<()> {
    let oscillators: Vec<_> = forest.oscillators().collect();
    sink.write_seq_len(oscillators.len() as u64)?;
    for osc in oscillators {
        sink.write_u64(forest.unit_position(osc)?)?;
    }
    let domains: Vec<_> = forest.domains().collect();
    sink.write_seq_len(domains.len() as u64)?;
    for id in domains {
        sink.write_u64(forest.ticks(id)?)?;
    }
    Ok(())
}

/// Restore what [`save_clocks`] wrote.
fn load_clocks<'a>(forest: &mut ClockForest, src: &mut impl Source<'a>) -> Result<()> {
    let domains: Vec<_> = forest.domains().collect();
    let count = src.read_seq_len(8)? as usize;
    let oscillators: Vec<_> = forest.oscillators().collect();
    if count != oscillators.len() {
        return Err(Error::State(format!(
            "snapshot has {count} oscillators, this machine has {}",
            oscillators.len()
        )));
    }
    // Unit positions first: §4.2's tick counters are anchored to them, so
    // restoring in the other order rebases every counter onto the old front.
    for osc in oscillators {
        forest.restore_unit_position(osc, src.read_u64()?)?;
    }
    let count = src.read_seq_len(8)? as usize;
    if count != domains.len() {
        return Err(Error::State(format!(
            "snapshot has {count} clock domains, this machine has {}",
            domains.len()
        )));
    }
    for id in domains {
        forest.restore_ticks(id, src.read_u64()?)?;
    }
    Ok(())
}

/// Write the scheduler's own architectural state (§4.5).
///
/// Virtual time, every pending event in fire order, the tie-break sequence
/// counter and the round-robin cursor. Re-deriving the queue by asking devices
/// to re-register would lose sub-tick phase — a timer 40 cycles from firing
/// would come back a whole period from firing — so the queue is written
/// verbatim.
fn save_sched(sched: &Scheduler, sink: &mut impl Sink) -> Result<()> {
    let snapshot = sched.snapshot();
    sink.write_u128(snapshot.now.raw())?;
    sink.write_u64(snapshot.next_seq)?;
    sink.write_u64(snapshot.cursor as u64)?;
    sink.write_seq_len(snapshot.events.len() as u64)?;
    for event in &snapshot.events {
        sink.write_u128(event.time.raw())?;
        sink.write_u64(event.id.seq())?;
        sink.write_u32(event.target.0)?;
        sink.write_u64(event.token)?;
    }
    Ok(())
}

/// Restore what [`save_sched`] wrote.
fn load_sched<'a>(sched: &mut Scheduler, src: &mut impl Source<'a>) -> Result<()> {
    let now = GlobalTime::from_raw(src.read_u128()?);
    let next_seq = src.read_u64()?;
    let cursor = usize::try_from(src.read_u64()?)
        .map_err(|_| Error::State(String::from("scheduler cursor does not fit this host")))?;
    // Sixteen bytes of instant, eight of sequence, four of target, eight of
    // token: an event cannot encode in fewer, so a corrupt count is caught
    // before anything is reserved.
    let count = src.read_seq_len(36)? as usize;
    let mut events = Vec::with_capacity(count.min(src.remaining()));
    for _ in 0..count {
        events.push(Event {
            time: GlobalTime::from_raw(src.read_u128()?),
            id: EventId::from_seq(src.read_u64()?),
            target: EventTarget(src.read_u32()?),
            token: src.read_u64()?,
        });
    }
    sched.restore(&SchedulerSnapshot {
        now,
        next_seq,
        cursor,
        events,
    })?;
    Ok(())
}

/// Write each net's per-source levels.
fn save_wires(nets: &[Net], sink: &mut impl Sink) -> Result<()> {
    sink.write_seq_len(nets.len() as u64)?;
    for net in nets {
        let levels = net.wire.snapshot();
        sink.write_seq_len(levels.len() as u64)?;
        for (id, level) in levels {
            sink.write_u64(id.raw())?;
            sink.write_u8(u8::from(level.is_high()))?;
        }
    }
    Ok(())
}

/// Restore what [`save_wires`] wrote.
fn load_wires<'a>(nets: &[Net], src: &mut impl Source<'a>) -> Result<()> {
    // A net encodes at least its own source count.
    let count = src.read_seq_len(8)? as usize;
    if count != nets.len() {
        return Err(Error::State(format!(
            "snapshot has {count} wire nets, this machine has {}",
            nets.len()
        )));
    }
    for net in nets {
        // Nine bytes per source: a `u64` id and a level byte.
        let sources = src.read_seq_len(9)? as usize;
        let mut levels = Vec::with_capacity(sources.min(src.remaining()));
        for _ in 0..sources {
            let id = WireId::new(src.read_u64()?);
            let level = match src.read_u8()? {
                0 => Level::Low,
                1 => Level::High,
                other => {
                    return Err(Error::State(format!("wire level {other} is not 0 or 1")));
                }
            };
            levels.push((id, level));
        }
        net.wire.restore(&levels);
    }
    // Restoring a level does not *deliver* it, and a sink's own fan-in is
    // derived state that nothing else rebuilds — so a re-announce is what
    // makes the sinks agree with the wires again. It has to come after every
    // net is restored: a sink that re-drives its output would otherwise write
    // over a net whose saved levels had not been put back yet.
    for net in nets {
        net.wire.refresh();
    }
    Ok(())
}

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

    #[test]
    fn the_machine_chunk_paths_cannot_collide_with_a_device() {
        // Object names come from the resolver's identifier grammar, which has
        // no `/`. That is the whole reason the machine's own chunks are spelled
        // with one.
        assert!(CLOCK_PATH.starts_with('/'));
        assert!(WIRE_PATH.starts_with('/'));
    }

    #[test]
    fn the_state_hash_is_a_function_of_the_bytes() {
        assert_eq!(fnv1a(b"abc"), fnv1a(b"abc"));
        assert_ne!(fnv1a(b"abc"), fnv1a(b"abd"));
        // An empty snapshot still hashes to the FNV offset basis rather than 0,
        // so "no state" and "hash not computed" are distinguishable.
        assert_eq!(fnv1a(b""), 0xcbf2_9ce4_8422_2325);
    }
}