use consortium_core::{
Region, TransportError, TryError, TryRecvTransport, TrySendTransport,
descriptor::DescriptorHeaderError,
};
use crate::{
descriptor::{
DESCRIPTOR_SIZE, Descriptor, FLAGS_OFFSET, Flags, MAGIC, MAX_FRAME_LEN, ReadOutcome,
VERSION,
},
packet::PACKET_HEADER_SIZE,
utils::wv_u16,
};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RingError {
MagicMismatch,
VersionMismatch,
CapacityMismatch {
descriptor: u16,
mapped: usize,
},
PointerOutOfBounds {
ptr: u16,
capacity: u16,
},
RegionLength {
len: usize,
},
FrameTooLarge {
len: usize,
max: usize,
},
BufferTooSmall {
len: usize,
needed: usize,
},
Faulted,
NonCoherentRegion,
}
impl core::fmt::Display for RingError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::MagicMismatch => f.write_str("CDBG magic mismatch"),
Self::VersionMismatch => f.write_str("CDBG version mismatch"),
Self::CapacityMismatch { descriptor, mapped } => write!(
f,
"descriptor capacity {descriptor} exceeds mapped length {mapped}"
),
Self::PointerOutOfBounds { ptr, capacity } => write!(
f,
"ring pointer {ptr} is out of bounds (capacity {capacity})"
),
Self::RegionLength { len } => write!(f, "a {len}-byte region cannot hold a ring"),
Self::FrameTooLarge { len, max } => {
write!(f, "a {len}-byte frame exceeds the ring's {max}-byte limit")
}
Self::BufferTooSmall { len, needed } => {
write!(f, "a {needed}-byte frame does not fit a {len}-byte buffer")
}
Self::Faulted => f.write_str("the ring writer latched an error"),
Self::NonCoherentRegion => f.write_str("a CDBG ring needs a cache-coherent region"),
}
}
}
impl core::error::Error for RingError {}
fn require_coherent<R: Region>(region: &R) -> Result<(), RingError> {
if region.is_coherent() {
Ok(())
} else {
Err(RingError::NonCoherentRegion)
}
}
pub(crate) fn validate<R: Region>(region: &R) -> Result<Descriptor, RingError> {
require_coherent(region)?;
let len = region.len();
if len < DESCRIPTOR_SIZE {
return Err(RingError::CapacityMismatch {
descriptor: 0,
mapped: len,
});
}
let desc = unsafe { Descriptor::from_raw_ptr(base(region)) };
desc.header.check(MAGIC, VERSION).map_err(|e| match e {
DescriptorHeaderError::BadVersion(..) => RingError::VersionMismatch,
_ => RingError::MagicMismatch,
})?;
if desc.capacity as usize > len {
return Err(RingError::CapacityMismatch {
descriptor: desc.capacity,
mapped: len,
});
}
let data_start = DESCRIPTOR_SIZE as u16;
for ptr in [desc.read_ptr, desc.write_ptr] {
if ptr < data_start || ptr >= desc.capacity {
return Err(RingError::PointerOutOfBounds {
ptr,
capacity: desc.capacity,
});
}
}
Ok(desc)
}
fn base<R: Region>(region: &R) -> *mut u8 {
unsafe { region.as_ptr() }
}
fn max_frame(capacity: u16) -> usize {
(capacity as usize)
.saturating_sub(DESCRIPTOR_SIZE + PACKET_HEADER_SIZE)
.min(MAX_FRAME_LEN)
}
pub struct RingWriter<R> {
region: R,
capacity: u16,
}
impl<R: Region> RingWriter<R> {
pub unsafe fn init(region: R) -> Result<Self, RingError> {
require_coherent(®ion)?;
let len = region.len();
let Ok(capacity) = u16::try_from(len) else {
return Err(RingError::RegionLength { len });
};
if max_frame(capacity) == 0 {
return Err(RingError::RegionLength { len });
}
let base = base(®ion);
for offset in DESCRIPTOR_SIZE..len {
#[allow(clippy::multiple_unsafe_ops_per_block)]
unsafe {
base.add(offset).write_volatile(0)
};
}
unsafe { Descriptor::new(capacity).write_to_raw_ptr(base) };
Ok(Self { region, capacity })
}
pub unsafe fn attach(region: R) -> Result<Self, RingError> {
let capacity = validate(®ion)?.capacity;
Ok(Self { region, capacity })
}
#[cfg(feature = "logger")]
pub(crate) unsafe fn attach_unchecked(region: R, capacity: u16) -> Self {
Self { region, capacity }
}
pub fn into_region(self) -> R {
self.region
}
pub fn snapshot(&self) -> Descriptor {
unsafe { Descriptor::from_raw_ptr(base(&self.region)) }
}
pub fn mark_panic(&mut self) {
let mut desc = self.snapshot();
desc.flags.insert(Flags::Panic);
unsafe { wv_u16(base(&self.region), FLAGS_OFFSET, desc.flags.bits()) };
}
}
impl<R: Region> TransportError for RingWriter<R> {
type Error = RingError;
}
impl<R: Region> TrySendTransport for RingWriter<R> {
fn try_send(&mut self, frame: &[u8]) -> Result<(), TryError<RingError>> {
let base = base(&self.region);
let mut desc = unsafe { Descriptor::from_raw_ptr(base) };
if desc.flags.contains(Flags::Error) {
return Err(TryError::Fail(RingError::Faulted));
}
if desc.capacity != self.capacity {
return Err(TryError::Fail(RingError::CapacityMismatch {
descriptor: desc.capacity,
mapped: self.region.len(),
}));
}
let wrote = unsafe { desc.write_one_frame(base, frame) };
if wrote {
return Ok(());
}
let max = self.max_send_size();
if frame.len() > max {
Err(TryError::Fail(RingError::FrameTooLarge {
len: frame.len(),
max,
}))
} else {
Err(TryError::WouldBlock)
}
}
fn max_send_size(&self) -> usize {
max_frame(self.capacity)
}
}
pub struct RingReader<R> {
region: R,
}
impl<R: Region> RingReader<R> {
pub unsafe fn new(region: R) -> Result<Self, RingError> {
validate(®ion)?;
Ok(Self { region })
}
pub fn into_region(self) -> R {
self.region
}
pub fn region(&self) -> &R {
&self.region
}
pub fn snapshot(&self) -> Descriptor {
unsafe { Descriptor::from_raw_ptr(base(&self.region)) }
}
fn checked_snapshot(&self) -> Result<Descriptor, RingError> {
let desc = self.snapshot();
if desc.capacity as usize > self.region.len() {
return Err(RingError::CapacityMismatch {
descriptor: desc.capacity,
mapped: self.region.len(),
});
}
if (desc.read_ptr as usize) < DESCRIPTOR_SIZE || desc.read_ptr >= desc.capacity {
return Err(RingError::PointerOutOfBounds {
ptr: desc.read_ptr,
capacity: desc.capacity,
});
}
Ok(desc)
}
pub fn drain(&mut self, mut on_frame: impl FnMut(&[u8])) -> Result<usize, RingError> {
let mut desc = self.checked_snapshot()?;
let mut buf = [0u8; MAX_FRAME_LEN];
let mut frames = 0;
loop {
match unsafe { desc.try_read_frame(base(&self.region), &mut buf) } {
ReadOutcome::Empty => return Ok(frames),
ReadOutcome::Frame(len) => {
frames += 1;
on_frame(&buf[..len]);
}
ReadOutcome::TooSmall { needed } => {
return Err(RingError::BufferTooSmall {
len: buf.len(),
needed,
});
}
}
}
}
}
impl<R: Region> TransportError for RingReader<R> {
type Error = RingError;
}
impl<R: Region> TryRecvTransport for RingReader<R> {
fn try_recv(&mut self, buf: &mut [u8]) -> Result<usize, TryError<RingError>> {
let mut desc = self.checked_snapshot()?;
match unsafe { desc.try_read_frame(base(&self.region), buf) } {
ReadOutcome::Empty => Err(TryError::WouldBlock),
ReadOutcome::Frame(len) => Ok(len),
ReadOutcome::TooSmall { needed } => Err(TryError::Fail(RingError::BufferTooSmall {
len: buf.len(),
needed,
})),
}
}
fn max_recv_size(&self) -> usize {
MAX_FRAME_LEN
}
}