From 1dadce2bd0c998f9674b8657f246b789a29de906 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Onur=20=C3=96zkan?= Date: Sun, 19 Jul 2026 22:39:55 +0300 Subject: [PATCH] libs/kaptan-tx: add bounded transmit queue --- libs/kaptan-tx/Cargo.toml | 10 ++ libs/kaptan-tx/src/lib.rs | 223 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 233 insertions(+) create mode 100644 libs/kaptan-tx/Cargo.toml create mode 100644 libs/kaptan-tx/src/lib.rs diff --git a/libs/kaptan-tx/Cargo.toml b/libs/kaptan-tx/Cargo.toml new file mode 100644 index 0000000..2ce4409 --- /dev/null +++ b/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 diff --git a/libs/kaptan-tx/src/lib.rs b/libs/kaptan-tx/src/lib.rs new file mode 100644 index 0000000..ba34674 --- /dev/null +++ b/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 { + /// 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 { + pending: [Option>; N], + reserved: Option>, + active: Option>, +} + +impl TransmitQueue { + /// 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) -> 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( + &mut self, + start: impl FnOnce(&Frame) -> Result<(), Error>, + ) -> Result { + 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> { + self.active.take() + } + + /// Releases the active frame after a terminal DMA fault. + pub fn fail(&mut self) -> Option> { + 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> { + 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 Default for TransmitQueue { + 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::::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::::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); + } +}