libs/kaptan-tx: add bounded transmit queue

libs/kaptan-tx/Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "kaptan-tx"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[lints]
+workspace = true

libs/kaptan-tx/src/lib.rs

@@ -0,0 +1,223 @@
+//! Bounded protocol-neutral transmit admission and ownership lifecycle.
+//!
+//! Protocol encoders own frame bytes. This crate owns only bounded priority
+//! admission and the transition from pending frame to active DMA ownership.
+
+#![no_std]
+
+/// Delivery importance used for bounded admission and selection.
+#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
+pub enum FrameClass {
+    /// Liveness or status traffic that best-effort samples cannot displace.
+    Critical,
+    /// Replaceable sample traffic where newest pending data is most useful.
+    BestEffort,
+}
+
+/// An owned protocol frame and its delivery class.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct PendingFrame<Frame> {
+    /// Queue admission and selection priority.
+    pub class: FrameClass,
+    /// Protocol-owned encoded payload.
+    pub frame: Frame,
+}
+
+/// Outcome of bounded queue admission.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum EnqueueResult {
+    /// The frame used an empty pending slot.
+    Enqueued,
+    /// The frame displaced one pending best-effort frame.
+    ReplacedBestEffort,
+    /// Critical admission failed because no best-effort pending frame existed.
+    CriticalQueueFull,
+    /// Best-effort admission failed because no replaceable pending frame existed.
+    BestEffortDropped,
+}
+
+/// Result of attempting to begin a transfer.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum StartOutcome {
+    /// No pending frame was available.
+    Idle,
+    /// The engine accepted a frame and owns it until a terminal outcome.
+    Started,
+}
+
+/// Bounded admission and exactly-one-active-transfer state.
+///
+/// Pending best-effort traffic is replaceable; active traffic is never
+/// replaceable. A selected frame moves into its final active storage before
+/// the engine sees its address. A rejected start moves it back to the
+/// reservation, so a failed submission cannot manufacture a false in-flight
+/// frame or discard the selected frame.
+pub struct TransmitQueue<Frame, const N: usize> {
+    pending: [Option<PendingFrame<Frame>>; N],
+    reserved: Option<PendingFrame<Frame>>,
+    active: Option<PendingFrame<Frame>>,
+}
+
+impl<Frame: Copy, const N: usize> TransmitQueue<Frame, N> {
+    /// Creates an empty transmit queue.
+    pub const fn new() -> Self {
+        Self {
+            pending: [None; N],
+            reserved: None,
+            active: None,
+        }
+    }
+
+    /// Admits one frame according to its delivery class.
+    pub fn enqueue(&mut self, frame: PendingFrame<Frame>) -> EnqueueResult {
+        if let Some(slot) = self.pending.iter_mut().find(|slot| slot.is_none()) {
+            *slot = Some(frame);
+            return EnqueueResult::Enqueued;
+        }
+
+        if let Some(slot) = self
+            .pending
+            .iter_mut()
+            .find(|slot| slot.is_some_and(|pending| pending.class == FrameClass::BestEffort))
+        {
+            *slot = Some(frame);
+            return EnqueueResult::ReplacedBestEffort;
+        }
+
+        match frame.class {
+            FrameClass::Critical => EnqueueResult::CriticalQueueFull,
+            FrameClass::BestEffort => EnqueueResult::BestEffortDropped,
+        }
+    }
+
+    /// Selects and starts at most one pending frame.
+    pub fn start_with<Error>(
+        &mut self,
+        start: impl FnOnce(&Frame) -> Result<(), Error>,
+    ) -> Result<StartOutcome, Error> {
+        if self.active.is_some() {
+            return Ok(StartOutcome::Idle);
+        }
+        if self.reserved.is_none() {
+            self.reserved = self.take_next();
+        }
+        self.active = self.reserved.take();
+
+        let Some(active) = self.active.as_ref() else {
+            return Ok(StartOutcome::Idle);
+        };
+        if let Err(error) = start(&active.frame) {
+            self.reserved = self.active.take();
+            return Err(error);
+        }
+
+        self.reserved = None;
+        Ok(StartOutcome::Started)
+    }
+
+    /// Releases the active frame after a terminal DMA completion.
+    pub fn complete(&mut self) -> Option<PendingFrame<Frame>> {
+        self.active.take()
+    }
+
+    /// Releases the active frame after a terminal DMA fault.
+    pub fn fail(&mut self) -> Option<PendingFrame<Frame>> {
+        self.active.take()
+    }
+
+    /// Returns whether an engine owns a frame.
+    pub const fn is_active(&self) -> bool {
+        self.active.is_some()
+    }
+
+    fn take_next(&mut self) -> Option<PendingFrame<Frame>> {
+        let critical = self
+            .pending
+            .iter()
+            .position(|slot| slot.is_some_and(|pending| pending.class == FrameClass::Critical));
+        let index = critical.or_else(|| self.pending.iter().position(Option::is_some));
+        index.and_then(|index| self.pending[index].take())
+    }
+}
+
+impl<Frame: Copy, const N: usize> Default for TransmitQueue<Frame, N> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{EnqueueResult, FrameClass, PendingFrame, StartOutcome, TransmitQueue};
+
+    #[test]
+    fn keeps_active_frame_while_replacing_pending_best_effort() {
+        let mut queue = TransmitQueue::<u8, 1>::new();
+        assert_eq!(
+            queue.enqueue(PendingFrame {
+                class: FrameClass::BestEffort,
+                frame: 1,
+            }),
+            EnqueueResult::Enqueued
+        );
+        assert_eq!(
+            queue.start_with(|_| Ok::<_, ()>(())),
+            Ok(StartOutcome::Started)
+        );
+        assert!(queue.is_active());
+
+        assert_eq!(
+            queue.enqueue(PendingFrame {
+                class: FrameClass::BestEffort,
+                frame: 2,
+            }),
+            EnqueueResult::Enqueued
+        );
+        assert_eq!(queue.complete().unwrap().frame, 1);
+    }
+
+    #[test]
+    fn rejects_start_without_claiming_active_ownership() {
+        let mut queue = TransmitQueue::<u8, 1>::new();
+        queue.enqueue(PendingFrame {
+            class: FrameClass::BestEffort,
+            frame: 7,
+        });
+
+        assert_eq!(queue.start_with(|_| Err::<(), _>(1)), Err(1));
+        assert!(!queue.is_active());
+        assert_eq!(
+            queue.start_with(|frame| {
+                assert_eq!(*frame, 7);
+                Ok::<_, ()>(())
+            }),
+            Ok(StartOutcome::Started)
+        );
+    }
+
+    #[test]
+    fn keeps_the_dma_source_at_its_active_address() {
+        let mut queue = TransmitQueue::<[u8; 2], 1>::new();
+        queue.enqueue(PendingFrame {
+            class: FrameClass::BestEffort,
+            frame: [1, 2],
+        });
+
+        let mut submitted_address = 0;
+        assert_eq!(
+            queue.start_with(|frame| {
+                submitted_address = frame.as_ptr() as usize;
+                Ok::<_, ()>(())
+            }),
+            Ok(StartOutcome::Started)
+        );
+        let active_address = queue
+            .active
+            .as_ref()
+            .expect("successful start must retain the active frame")
+            .frame
+            .as_ptr() as usize;
+
+        assert_eq!(submitted_address, active_address);
+    }
+}